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