From e5d8c17bb46658e13990f72bf50ff7bbf1fbf99a Mon Sep 17 00:00:00 2001
From: Rosa Gutierrez <rosa@37signals.com>
Date: Wed, 29 Jul 2026 21:23:01 +0200
Subject: [PATCH 1/3] Bump solid_queue to 1.5.1

---
 Gemfile.lock               | 2 +-
 lib/solid_queue/version.rb | 2 +-
 2 files changed, 2 insertions(+), 2 deletions(-)

diff --git a/Gemfile.lock b/Gemfile.lock
index 20ce2779..7ef8b851 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/lib/solid_queue/version.rb b/lib/solid_queue/version.rb
index 967d4cc4..bb2ee0ed 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

From 78b1c4a8dd3e8ec22338172656a992661f4e0a4e Mon Sep 17 00:00:00 2001
From: JP Camara <jp@jpcamara.com>
Date: Wed, 29 Jul 2026 22:12:57 -0400
Subject: [PATCH 2/3] 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 a cursor-free
discovery pass on a monotonic-clock cadence seeds and heals the
position. The cadence (claim_cursors_discovery_interval, one second by
default) bounds both how often an idle worker pays the graveyard scan
and how long it takes to notice brand-new work; the test environment
shortens it so latency-sensitive integration suites keep classic pickup
timing. The cursor is not a lower bound for live
work (a lower-id claim can roll back after we advance past it, and
sequences are not commit-ordered), so discovery is fundamental to
finding work, and polling is graveyard-resistant rather than immune:
discovery still scans the graveyard once per cadence.

Claim ordering within a priority changes from enqueue order (job_id) to
readiness order (id), matching the cursor's index position; the polling
indexes change accordingly and the README includes the migration. A
strictly-higher-priority arrival waits for the next discovery, inside
the slack SKIP LOCKED already implies. Row-constructor seeks and the
motivating pathology are PostgreSQL-specific, so other adapters keep
the classic path; config.solid_queue.claim_cursors = false disables.

Pinned-horizon bench (200k dead tuples): 9.36 -> 0.20 ms per idle poll.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 README.md                                     |  74 ++++++
 app/models/solid_queue/queue_selector.rb      |  14 +-
 app/models/solid_queue/ready_execution.rb     |  83 +++++-
 .../ready_execution/claim_cursors.rb          |  77 ++++++
 .../install/templates/db/queue_schema.rb      |   2 +
 lib/solid_queue.rb                            |   2 +
 test/dummy/config/environments/test.rb        |   4 +
 test/dummy/db/queue_schema.rb                 |   2 +
 test/models/solid_queue/claim_cursors_test.rb | 250 ++++++++++++++++++
 test/test_helper.rb                           |   3 +
 10 files changed, 499 insertions(+), 12 deletions(-)
 create mode 100644 app/models/solid_queue/ready_execution/claim_cursors.rb
 create mode 100644 test/models/solid_queue/claim_cursors_test.rb

diff --git a/README.md b/README.md
index 700dec3d..66050afa 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)
@@ -431,6 +432,79 @@ 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. A cursor-free
+discovery query is therefore
+a fundamental part of finding work, not merely defensive healing. It runs on a time-based
+cadence (`config.solid_queue.claim_cursors_discovery_interval`, 1 second by default), seeds
+or heals the position, and discovers higher-priority arrivals below it. The interval bounds
+how long an idle worker takes to notice brand-new work, and how often it pays the graveyard
+scan: lower it for snappier pickup, raise it to spend even less CPU while a horizon is
+pinned.
+
+Discovery still scans the dead-tuple graveyard at that cadence, so cursors reduce and
+amortize the cost rather than eliminate it. 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:
diff --git a/app/models/solid_queue/queue_selector.rb b/app/models/solid_queue/queue_selector.rb
index d0d61dd1..bba3a0bb 100644
--- a/app/models/solid_queue/queue_selector.rb
+++ b/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?
diff --git a/app/models/solid_queue/ready_execution.rb b/app/models/solid_queue/ready_execution.rb
index ea14c193..14cd0065 100644
--- a/app/models/solid_queue/ready_execution.rb
+++ b/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,79 @@ def aggregated_count_across(queue_list)
         QueueSelector.new(queue_list, self).scoped_relations.map(&:count).sum
       end
 
+      def claim_cursors
+        @claim_cursors ||= ClaimCursors.new
+      end
+
       private
-        def select_and_lock(queue_relation, process_id, limit)
+        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_due, position = claim_cursors.state(key)
+
+          if discovery_due
+            claim_discovering(key, queue_relation, process_id, limit)
+          elsif position
+            claim_along_cursor(key, position, queue_relation, process_id, limit)
+          else
+            []
           end
         end
 
-        def select_candidates(queue_relation, limit)
+        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. Discovery both seeds
+        # the fast path and finds rows that appeared below its position.
+        def claim_discovering(key, queue_relation, process_id, limit)
+          candidates, claimed = claim_candidates(queue_relation, process_id, limit)
+          claim_cursors.record_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
+
+        # 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 +107,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 00000000..86a9816a
--- /dev/null
+++ b/app/models/solid_queue/ready_execution/claim_cursors.rb
@@ -0,0 +1,77 @@
+# 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. The periodic cursor-free discovery query is therefore
+    # fundamental to finding work, not merely defensive healing.
+    class ClaimCursors
+      def initialize(clock: -> { ::Process.clock_gettime(::Process::CLOCK_MONOTONIC) })
+        @clock = clock
+        @mutex = Mutex.new
+        @positions = {}
+        @next_discovery_at = {}
+      end
+
+      def state(key)
+        @mutex.synchronize { [ discovery_due_without_lock?(key), @positions[key]&.dup ] }
+      end
+
+      def position(key)
+        @mutex.synchronize { @positions[key]&.dup }
+      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 { discovery_due_without_lock?(key) }
+      end
+
+      def record_discovery(key, position)
+        @mutex.synchronize do
+          @positions[key] = position&.dup
+          @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 reset!
+        @mutex.synchronize do
+          @positions.clear
+          @next_discovery_at.clear
+        end
+      end
+
+      private
+        def discovery_due_without_lock?(key)
+          deadline = @next_discovery_at[key]
+          deadline.nil? || monotonic_time >= deadline
+        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 85194b6a..8694354a 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,9 @@
     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"
   end
 
   create_table "solid_queue_recurring_executions", force: :cascade do |t|
diff --git a/lib/solid_queue.rb b/lib/solid_queue.rb
index 654a6be6..4aa6609f 100644
--- a/lib/solid_queue.rb
+++ b/lib/solid_queue.rb
@@ -26,6 +26,8 @@ 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 :process_heartbeat_interval, default: 60.seconds
   mattr_accessor :process_alive_threshold, default: 5.minutes
diff --git a/test/dummy/config/environments/test.rb b/test/dummy/config/environments/test.rb
index 4a07d140..a965e3c7 100644
--- a/test/dummy/config/environments/test.rb
+++ b/test/dummy/config/environments/test.rb
@@ -60,6 +60,10 @@
 
   config.solid_queue.shutdown_timeout = 2.seconds
 
+  # Keep job pickup snappy for latency-sensitive integration tests; cursor
+  # cadence behavior is pinned deterministically in ClaimCursorsTest
+  config.solid_queue.claim_cursors_discovery_interval = 0.05.seconds
+
   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 697c2e92..ccce8e25 100644
--- a/test/dummy/db/queue_schema.rb
+++ b/test/dummy/db/queue_schema.rb
@@ -83,7 +83,9 @@
     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"
   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_test.rb b/test/models/solid_queue/claim_cursors_test.rb
new file mode 100644
index 00000000..42d12247
--- /dev/null
+++ b/test/models/solid_queue/claim_cursors_test.rb
@@ -0,0 +1,250 @@
+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
+    SolidQueue.claim_cursors_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
+  end
+
+  test "advance is monotonic per key" do
+    cursors.advance("*", [ 0, 10 ])
+    cursors.advance("*", [ 0, 5 ])
+    assert_equal [ 0, 10 ], cursors.position("*")
+
+    cursors.advance("*", [ 1, 1 ])
+    assert_equal [ 1, 1 ], cursors.position("*")
+  end
+
+  test "clear only removes the position it observed" do
+    cursors.advance("*", [ 0, 10 ])
+
+    cursors.clear("*", [ 0, 5 ]) # stale observation from a slower thread
+    assert_equal [ 0, 10 ], cursors.position("*")
+
+    cursors.clear("*", [ 0, 10 ])
+    assert_nil cursors.position("*")
+  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("*")
+
+    assert_equal 1, claim(1).size
+    assert_equal position_of(executions.second), cursors.position("*")
+  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("*")
+    assert_not cursors.discovery_due?("*")
+
+    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("*")
+
+    cursors.expire_discovery!("*")
+    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("*", [ execution.priority, execution.id + 1000 ])
+
+    assert_empty claim(1)
+    assert_nil cursors.position("*")
+
+    cursors.expire_discovery!("*")
+    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 "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("*")
+    assert cursors.discovery_due?("*")
+  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 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("*")
+
+    peer.execute("ROLLBACK")
+
+    SolidQueue::ReadyExecution.claim_cursors.expire_discovery!("*")
+    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 99d00ba2..c86e27a0 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

From ed3817880aff4f00629a53b5ce5e4cab52ea79c6 Mon Sep 17 00:00:00 2001
From: JP Camara <jp@jpcamara.com>
Date: Thu, 30 Jul 2026 01:44:34 -0400
Subject: [PATCH 3/3] Bound the claim cursor discovery pass with a sequence
 watermark

Claim cursors made fast polls cheap, but the cursor-free discovery pass
that keeps them correct still scans from the low end of the polling index,
so it costs O(graveyard) and stays the one term that grows without bound
while the xmin horizon is pinned. On the system that motivated cursors it
was 202,589 buffer hits per pass against 3,442 for a fast poll: roughly
77% of what polling still cost.

Discovery now comes in two forms. A floored pass runs on the existing
short cadence, bounded below by the 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` still finds a
strictly-higher-priority arrival without rescanning the graveyard buried
underneath it. The floor is a non-leading column of (priority, id) and so
cannot be a boundary, but PostgreSQL still applies it inside the index and
skips the heap visits that make the unbounded pass expensive. An unbounded
pass runs on a new long cadence (claim_cursors_full_discovery_interval, 10
seconds by default), records the next floor, and remains the only pass
that reaches a row sitting at or below the floor.

Two rules keep the floored pass sound. It claims only the region below the
cursor while the fast seek claims the region above it in the same poll, so
the two stay disjoint and claims keep their (priority, id) order. And it
never advances or clears the cursor -- it may only seed one an empty seek
cleared -- because a pass that saw only part of the index would otherwise
move the cursor over rows its floor had hidden, stranding them below it.
Claiming in (priority, id) order is what makes that possible: the highest
id a pass claims is not an id it has swept past, so a watermark that
followed the claims would jump over unclaimed lower ids at other
priorities.

The floor's guarantee leans on ids being handed out in order, so it is
recorded only where that holds: PostgreSQL 10+, whose catalog exposes the
sequence's cache setting, with the cache at its default of 1. Anything
else -- an older server, CACHE above 1, a table without its own sequence
-- records no floor and keeps every pass unbounded. The full deadline is
jittered only downward, so processes that booted together spread their
graveyard scans without the interval ever ceasing to be an upper bound,
and cursor state lives in a registry keyed by connection pool: floors and
deadlines are positions in one table's id space and must not leak between
datastores a process might poll.

Floored claims also run with sort plans off (SET LOCAL enable_sort and,
where it exists, enable_incremental_sort). With no visible rows and fresh
statistics -- the pinned-horizon state itself -- 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 sort can never beat a polling
index that already provides the order. The settings live inside a
savepoint, restored before it releases and reverted by PostgreSQL if it
rolls back, so they cannot escape a caller's transaction.

The tradeoff is healing latency. A rolled-back competing claim and a
commit that lands after the floor was read both leave a row that only an
unbounded pass can reach, and those now wait up to the full interval plus
the short cadence it piggybacks on, rather than the short interval alone.
That latency is a fixed, configurable constant rather than something that
grows with the graveyard, which is why this is preferable to sweeping the
graveyard incrementally. Setting the interval to 0 makes every discovery
pass unbounded again.

Pinned-horizon bench (200k unreapable dead tuples, heap decorrelated from
id order as a long-lived table's is): discovery pass 200,283 -> 982 buffer
hits and 49.1 -> 9.8 ms; a 200-poll loop at Solid Queue's default cadences
4,005,860 -> 418,406 buffer hits and 1,009 -> 360 ms.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 README.md                                     |  45 +++-
 app/models/solid_queue/ready_execution.rb     | 109 ++++++++-
 .../ready_execution/claim_cursors.rb          |  68 +++++-
 lib/solid_queue.rb                            |   1 +
 test/dummy/config/environments/test.rb        |   7 +-
 test/models/solid_queue/claim_cursors_test.rb | 229 ++++++++++++++++++
 6 files changed, 428 insertions(+), 31 deletions(-)

diff --git a/README.md b/README.md
index 66050afa..612854d8 100644
--- a/README.md
+++ b/README.md
@@ -448,18 +448,39 @@ 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. A cursor-free
-discovery query is therefore
-a fundamental part of finding work, not merely defensive healing. It runs on a time-based
-cadence (`config.solid_queue.claim_cursors_discovery_interval`, 1 second by default), seeds
-or heals the position, and discovers higher-priority arrivals below it. The interval bounds
-how long an idle worker takes to notice brand-new work, and how often it pays the graveyard
-scan: lower it for snappier pickup, raise it to spend even less CPU while a horizon is
-pinned.
-
-Discovery still scans the dead-tuple graveyard at that cadence, so cursors reduce and
-amortize the cost rather than eliminate it. Polling is graveyard-resistant, not
-dead-tuple-immune.
+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
diff --git a/app/models/solid_queue/ready_execution.rb b/app/models/solid_queue/ready_execution.rb
index 14cd0065..7c1ef0c3 100644
--- a/app/models/solid_queue/ready_execution.rb
+++ b/app/models/solid_queue/ready_execution.rb
@@ -23,11 +23,20 @@ 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 ||= ClaimCursors.new
+        claim_cursors_registry.compute_if_absent(connection_pool) { ClaimCursors.new }
       end
 
       private
+        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
 
@@ -35,14 +44,13 @@ def select_and_lock(key, queue_relation, process_id, limit)
             return claim_classically(queue_relation, process_id, limit)
           end
 
-          discovery_due, position = claim_cursors.state(key)
+          discovery, floor, position = claim_cursors.state(key)
 
-          if discovery_due
-            claim_discovering(key, queue_relation, process_id, limit)
-          elsif position
-            claim_along_cursor(key, position, queue_relation, process_id, limit)
+          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
 
@@ -51,11 +59,54 @@ def claim_classically(queue_relation, process_id, limit)
           claimed
         end
 
-        # Cursor-free claim in full (priority, id) order. Discovery both seeds
-        # the fast path and finds rows that appeared below its position.
+        # 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_discovery(key, position_of(candidates.last))
+
+          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
 
@@ -87,6 +138,44 @@ def claim_candidates(relation, process_id, limit)
           [ 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?
diff --git a/app/models/solid_queue/ready_execution/claim_cursors.rb b/app/models/solid_queue/ready_execution/claim_cursors.rb
index 86a9816a..c85d78e2 100644
--- a/app/models/solid_queue/ready_execution/claim_cursors.rb
+++ b/app/models/solid_queue/ready_execution/claim_cursors.rb
@@ -10,24 +10,40 @@ class ReadyExecution
     # 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. The periodic cursor-free discovery query is therefore
-    # fundamental to finding work, not merely defensive healing.
+    # 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 = {}
         @next_discovery_at = {}
+        @next_full_discovery_at = {}
       end
 
       def state(key)
-        @mutex.synchronize { [ discovery_due_without_lock?(key), @positions[key]&.dup ] }
+        @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]
@@ -42,12 +58,29 @@ def clear(key, expected_position)
       end
 
       def discovery_due?(key)
-        @mutex.synchronize { discovery_due_without_lock?(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_discovery(key, position)
+      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 record_floored_discovery(key, seed_position = nil)
+        @mutex.synchronize do
+          @positions[key] = seed_position.dup if seed_position && @positions[key].nil?
           @next_discovery_at[key] = monotonic_time + SolidQueue.claim_cursors_discovery_interval
         end
       end
@@ -56,19 +89,40 @@ 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
           @next_discovery_at.clear
+          @next_full_discovery_at.clear
         end
       end
 
       private
-        def discovery_due_without_lock?(key)
-          deadline = @next_discovery_at[key]
+        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
diff --git a/lib/solid_queue.rb b/lib/solid_queue.rb
index 4aa6609f..02f4a347 100644
--- a/lib/solid_queue.rb
+++ b/lib/solid_queue.rb
@@ -28,6 +28,7 @@ module SolidQueue
   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/test/dummy/config/environments/test.rb b/test/dummy/config/environments/test.rb
index a965e3c7..10a4c80f 100644
--- a/test/dummy/config/environments/test.rb
+++ b/test/dummy/config/environments/test.rb
@@ -60,9 +60,12 @@
 
   config.solid_queue.shutdown_timeout = 2.seconds
 
-  # Keep job pickup snappy for latency-sensitive integration tests; cursor
-  # cadence behavior is pinned deterministically in ClaimCursorsTest
+  # 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")
diff --git a/test/models/solid_queue/claim_cursors_test.rb b/test/models/solid_queue/claim_cursors_test.rb
index 42d12247..c5c8065e 100644
--- a/test/models/solid_queue/claim_cursors_test.rb
+++ b/test/models/solid_queue/claim_cursors_test.rb
@@ -9,13 +9,16 @@ class SolidQueue::ClaimCursorsTest < ActiveSupport::TestCase
     # 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
@@ -157,6 +160,206 @@ class SolidQueue::ClaimCursorsTest < ActiveSupport::TestCase
     assert_equal [ immediate.job_id, scheduled.job_id ], claimed_jobs.map(&:active_job_id)
   end
 
+  test "discovery is unbounded until a pass has recorded a watermark" 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_not_nil cursors.floor("*")
+  end
+
+  test "a floored pass sweeps below the cursor and seeks above it in one poll" do
+    AddToBufferJob.perform_later("seed")
+    claim(1)
+
+    AddToBufferJob.perform_later("next")
+    cursors.expire_discovery!("*")
+
+    queries = capture_candidate_queries { assert_equal 1, claim(1).size }
+
+    assert_equal 2, queries.size
+    assert_includes queries.first, "(priority, id) <= ("
+    assert_includes queries.first, "id > "
+    assert_includes queries.second, "(priority, id) > ("
+  end
+
+  test "a floored pass finds a higher priority arrival without scanning below the watermark" do
+    AddToBufferJob.set(priority: 5).perform_later("seed")
+    claim(1)
+    position = cursors.position("*")
+
+    AddToBufferJob.set(priority: 1).perform_later("higher")
+    cursors.expire_discovery!("*")
+
+    queries = capture_candidate_queries do
+      claimed = claim(1)
+      assert_equal 1, SolidQueue::Job.find(claimed.sole.job_id).priority
+    end
+
+    assert_equal 1, queries.size
+    assert_includes queries.sole, "id > "
+    # 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("*")
+  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!("*")
+    assert_equal higher.job_id, claimed_active_job_ids(1).sole
+
+    cursors.expire_discovery!("*")
+    assert_equal lower.job_id, claimed_active_job_ids(1).sole
+  end
+
+  test "a floored pass with no cursor spans every priority and seeds one" do
+    AddToBufferJob.set(priority: 5).perform_later("seed")
+    claim(1)
+    assert_empty claim(1)
+    assert_nil cursors.position("*")
+
+    AddToBufferJob.set(priority: 1).perform_later("higher")
+    cursors.expire_discovery!("*")
+
+    queries = capture_candidate_queries { assert_equal 1, claim(1).size }
+
+    assert_equal 1, queries.size
+    assert_includes queries.sole, "id > "
+    assert_not_includes queries.sole, "(priority, id) <= ("
+    assert_not_nil cursors.position("*")
+  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("*")
+
+    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("*", seed_position, stranded.id)
+
+    cursors.expire_discovery!("*")
+    assert_empty claim(1)
+
+    cursors.expire_full_discovery!("*")
+    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?("*")
+    assert_not cursors.full_discovery_due?("*")
+
+    cursors.expire_discovery!("*")
+    assert cursors.discovery_due?("*")
+    assert_not cursors.full_discovery_due?("*")
+
+    AddToBufferJob.perform_later("next")
+    cursors.expire_full_discovery!("*")
+    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!("*")
+
+    queries = capture_candidate_queries { assert_equal 1, claim(1).size }
+
+    assert_equal 1, queries.size
+    assert_not_includes queries.sole, "id > "
+  end
+
+  test "a sequence cache above one records no floor and keeps discovery unbounded" do
+    with_sequence_cache(5) do
+      AddToBufferJob.perform_later("seed")
+      claim(1)
+      assert_nil cursors.floor("*")
+
+      AddToBufferJob.perform_later("next")
+      cursors.expire_discovery!("*")
+
+      queries = capture_candidate_queries { assert_equal 1, claim(1).size }
+      assert_not_includes queries.sole, "id > "
+    end
+  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("*", [ 0, 1 ], 1)
+
+      now += 0.7 * SolidQueue.claim_cursors_full_discovery_interval
+      assert_not clocked.full_discovery_due?("*")
+
+      now += 0.3 * SolidQueue.claim_cursors_full_discovery_interval
+      assert clocked.full_discovery_due?("*")
+    end
+  end
+
+  test "cursor state is scoped to the datastore" do
+    cursors.advance("*", [ 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("*")
+    other_cursors.advance("*", [ 0, 99 ])
+
+    SolidQueue::ReadyExecution.unstub(:connection_pool)
+    assert_equal [ 0, 10 ], cursors.position("*")
+  ensure
+    other_cursors&.reset!
+  end
+
+  test "a table without its own sequence records no floor instead of erroring" do
+    SolidQueue::ReadyExecution.stubs(:sequence_name).returns("nonexistent_sequence")
+    AddToBufferJob.perform_later("seed")
+
+    assert_equal 1, claim(1).size
+    assert_nil cursors.floor("*")
+  end
+
+  test "floored passes run with sort plans off and restore the setting" do
+    AddToBufferJob.perform_later("seed")
+    claim(1)
+
+    AddToBufferJob.perform_later("next")
+    cursors.expire_discovery!("*")
+
+    statements = capture_planner_settings { assert_equal 1, claim(1).size }
+
+    # On the graveyard's near-zero row estimates every plan ties, and the tie
+    # can land on a legacy job_id index that re-reads every dead tuple
+    assert_equal [ "SET LOCAL enable_sort = OFF", "SET LOCAL enable_sort = 'on'" ],
+      statements.grep(/enable_sort\b.*(OFF|on)/)
+    assert_equal "on", SolidQueue::Record.connection.select_value("SHOW enable_sort")
+  end
+
   test "disabling claim cursors uses the classic path without changing cursor state" do
     SolidQueue.claim_cursors = false
     AddToBufferJob.perform_later("classic")
@@ -182,6 +385,32 @@ def position_of(execution)
       [ execution.priority, execution.id ]
     end
 
+    def claimed_active_job_ids(limit)
+      claim(limit).map { |execution| SolidQueue::Job.find(execution.job_id).active_job_id }
+    end
+
+    def with_sequence_cache(cache)
+      connection = SolidQueue::Record.connection
+      sequence = connection.quote_table_name(SolidQueue::ReadyExecution.sequence_name)
+      connection.execute("ALTER SEQUENCE #{sequence} CACHE #{cache}")
+      yield
+    ensure
+      connection.execute("ALTER SEQUENCE #{sequence} CACHE 1")
+    end
+
+    def capture_planner_settings
+      statements = []
+      subscriber = ActiveSupport::Notifications.subscribe("sql.active_record") do |event|
+        sql = event.payload[:sql]
+        statements << sql if sql.start_with?("SET LOCAL enable_")
+      end
+
+      yield
+      statements
+    ensure
+      ActiveSupport::Notifications.unsubscribe(subscriber) if subscriber
+    end
+
     def capture_candidate_queries
       queries = []
       subscriber = ActiveSupport::Notifications.subscribe("sql.active_record") do |event|