diff --git a/Gemfile.lock b/Gemfile.lock
index 20ce27795..7ef8b8519 100644
--- a/Gemfile.lock
+++ b/Gemfile.lock
@@ -1,7 +1,7 @@
 PATH
   remote: .
   specs:
-    solid_queue (1.5.0)
+    solid_queue (1.5.1)
       activejob (>= 7.1)
       activerecord (>= 7.1)
       concurrent-ruby (>= 1.3.1)
diff --git a/README.md b/README.md
index 700dec3d2..612854d85 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,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:
diff --git a/app/models/solid_queue/queue_selector.rb b/app/models/solid_queue/queue_selector.rb
index d0d61dd14..bba3a0bb8 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 ea14c193a..7c1ef0c31 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,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)
diff --git a/app/models/solid_queue/ready_execution/claim_cursors.rb b/app/models/solid_queue/ready_execution/claim_cursors.rb
new file mode 100644
index 000000000..c85d78e20
--- /dev/null
+++ b/app/models/solid_queue/ready_execution/claim_cursors.rb
@@ -0,0 +1,131 @@
+# frozen_string_literal: true
+
+module SolidQueue
+  class ReadyExecution
+    # Process-local claim cursors: for each queue key, the last (priority, id)
+    # position this process observed. Cursor-guided queries can descend the
+    # polling index past dead tuples instead of scanning them.
+    #
+    # A cursor is not a lower bound for live work. A competing claim below it
+    # can roll back (its own transaction or an enclosing one), a row can
+    # receive a lower sequence id but commit after the cursor advances, and an
+    # empty SKIP LOCKED seek can mean the remaining rows were locked elsewhere
+    # rather than gone. Discovery is therefore fundamental to finding work, not
+    # merely defensive healing, and it comes in two forms. A floored pass runs
+    # on the short cadence, bounded below by the id watermark the last full
+    # pass recorded, which every row allocated since then exceeds. The
+    # unbounded full pass runs on the long cadence; it is the only pass that
+    # reaches rows at or below that watermark, and so the only one allowed to
+    # move a cursor.
+    class ClaimCursors
+      # Deadlines land up to this fraction BELOW the interval: worker processes
+      # that booted together don't scan the graveyard in lockstep, and the
+      # interval stays an upper bound on how long healing waits
+      FULL_DISCOVERY_JITTER = 0.25
+
+      def initialize(clock: -> { ::Process.clock_gettime(::Process::CLOCK_MONOTONIC) })
+        @clock = clock
+        @mutex = Mutex.new
+        @positions = {}
+        @floors = {}
+        @next_discovery_at = {}
+        @next_full_discovery_at = {}
+      end
+
+      def state(key)
+        @mutex.synchronize { [ discovery_without_lock(key), @floors[key], @positions[key]&.dup ] }
+      end
+
+      def position(key)
+        @mutex.synchronize { @positions[key]&.dup }
+      end
+
+      def floor(key)
+        @mutex.synchronize { @floors[key] }
+      end
+
+      def advance(key, position)
+        @mutex.synchronize do
+          current = @positions[key]
+          @positions[key] = position.dup if current.nil? || (current <=> position).negative?
+        end
+      end
+
+      def clear(key, expected_position)
+        @mutex.synchronize do
+          @positions.delete(key) if @positions[key] == expected_position
+        end
+      end
+
+      def discovery_due?(key)
+        @mutex.synchronize { due_without_lock?(@next_discovery_at, key) }
+      end
+
+      def full_discovery_due?(key)
+        @mutex.synchronize { due_without_lock?(@next_full_discovery_at, key) }
+      end
+
+      def record_full_discovery(key, position, floor)
+        @mutex.synchronize do
+          @positions[key] = position&.dup
+          @floors[key] = floor
+          @next_discovery_at[key] = monotonic_time + SolidQueue.claim_cursors_discovery_interval
+          @next_full_discovery_at[key] = monotonic_time + jittered_full_discovery_interval
+        end
+      end
+
+      # A floored pass only sees the part of the index above its floor, so it may
+      # seed a missing position but must never advance or clear one: the rows its
+      # floor hid would end up below the cursor, out of reach of every query but
+      # the next full pass.
+      def 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
+
+      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_without_lock(key)
+          return nil unless due_without_lock?(@next_discovery_at, key)
+          return :full if @floors[key].nil? || due_without_lock?(@next_full_discovery_at, key)
+
+          :floored
+        end
+
+        def due_without_lock?(deadlines, key)
+          deadline = deadlines[key]
+          deadline.nil? || monotonic_time >= deadline
+        end
+
+        def jittered_full_discovery_interval
+          interval = SolidQueue.claim_cursors_full_discovery_interval
+          interval - rand * FULL_DISCOVERY_JITTER * interval
+        end
+
+        def monotonic_time
+          @clock.call
+        end
+    end
+  end
+end
diff --git a/lib/generators/solid_queue/install/templates/db/queue_schema.rb b/lib/generators/solid_queue/install/templates/db/queue_schema.rb
index 85194b6a8..8694354a4 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 654a6be63..02f4a3476 100644
--- a/lib/solid_queue.rb
+++ b/lib/solid_queue.rb
@@ -26,6 +26,9 @@ module SolidQueue
   mattr_accessor :app_executor, :on_thread_error, :connects_to
 
   mattr_accessor :use_skip_locked, default: true
+  mattr_accessor :claim_cursors, default: true
+  mattr_accessor :claim_cursors_discovery_interval, default: 1.second
+  mattr_accessor :claim_cursors_full_discovery_interval, default: 10.seconds
 
   mattr_accessor :process_heartbeat_interval, default: 60.seconds
   mattr_accessor :process_alive_threshold, default: 5.minutes
diff --git a/lib/solid_queue/version.rb b/lib/solid_queue/version.rb
index 967d4cc48..bb2ee0ed7 100644
--- a/lib/solid_queue/version.rb
+++ b/lib/solid_queue/version.rb
@@ -1,5 +1,5 @@
 module SolidQueue
-  VERSION = "1.5.0"
+  VERSION = "1.5.1"
 
   def self.next_major_version
     Gem::Version.new(VERSION).segments.first + 1
diff --git a/test/dummy/config/environments/test.rb b/test/dummy/config/environments/test.rb
index 4a07d140a..10a4c80fe 100644
--- a/test/dummy/config/environments/test.rb
+++ b/test/dummy/config/environments/test.rb
@@ -60,6 +60,13 @@
 
   config.solid_queue.shutdown_timeout = 2.seconds
 
+  # Keep job pickup snappy for latency-sensitive integration tests, and keep the
+  # unbounded pass frequent enough that anything only it can heal heals inside a
+  # test's patience, while still leaving most passes floored. Cursor cadence
+  # behavior is pinned deterministically in ClaimCursorsTest
+  config.solid_queue.claim_cursors_discovery_interval = 0.05.seconds
+  config.solid_queue.claim_cursors_full_discovery_interval = 1.second
+
   config.log_formatter = proc do |severity, timestamp, progname, msg|
     ts = timestamp.getlocal.strftime("%H:%M:%S.%3N")
     "#{ts} #{msg}\n"
diff --git a/test/dummy/db/queue_schema.rb b/test/dummy/db/queue_schema.rb
index 697c2e928..ccce8e25b 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 000000000..c5c8065e0
--- /dev/null
+++ b/test/models/solid_queue/claim_cursors_test.rb
@@ -0,0 +1,479 @@
+require "test_helper"
+
+class SolidQueue::ClaimCursorsTest < ActiveSupport::TestCase
+  PROCESS_ID = 42
+
+  setup do
+    skip "Claim cursors require PostgreSQL" unless SolidQueue::Record.connection_db_config.adapter.match?(/postg/i)
+    SolidQueue::ReadyExecution.claim_cursors.reset!
+    # The dummy app shortens the interval for integration latency; these tests
+    # control discovery explicitly via expire_discovery!
+    @original_discovery_interval = SolidQueue.claim_cursors_discovery_interval
+    @original_full_discovery_interval = SolidQueue.claim_cursors_full_discovery_interval
+    SolidQueue.claim_cursors_discovery_interval = 10.minutes
+    SolidQueue.claim_cursors_full_discovery_interval = 10.minutes
+  end
+
+  teardown do
+    SolidQueue::ReadyExecution.claim_cursors.reset!
+    SolidQueue.claim_cursors = true
+    SolidQueue.claim_cursors_discovery_interval = @original_discovery_interval if @original_discovery_interval
+    SolidQueue.claim_cursors_full_discovery_interval = @original_full_discovery_interval if @original_full_discovery_interval
+  end
+
+  test "advance is monotonic per key" do
+    cursors.advance("*", [ 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 "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")
+
+    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 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|
+        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 99d00ba27..c86e27a08 100644
--- a/test/test_helper.rb
+++ b/test/test_helper.rb
@@ -46,6 +46,9 @@ def destroy_records
   end
 
   setup do
+    # Tests model fresh processes: claim cursors are process-local state that
+    # must not outlive the records they were derived from
+    SolidQueue::ReadyExecution.claim_cursors.reset!
     @_on_thread_error = SolidQueue.on_thread_error
     SolidQueue.on_thread_error = silent_on_thread_error_for(ExpectedTestError, @_on_thread_error)
     ActiveJob::QueueAdapters::SolidQueueAdapter.stopping = false