From 34909341a16ac5913e00fbbb51a0a7ed40008c8d Mon Sep 17 00:00:00 2001
From: BBrannick <bbrannick@gocardless.com>
Date: Thu, 18 Jun 2026 16:05:43 +0100
Subject: [PATCH 01/11] Add $3 run_at lower-bound parameter to lock_job and
 find_job_to_lock SQL queries

---
 lib/que/sql.rb | 3 +++
 1 file changed, 3 insertions(+)

diff --git a/lib/que/sql.rb b/lib/que/sql.rb
index 0de73b6c..449efcdc 100644
--- a/lib/que/sql.rb
+++ b/lib/que/sql.rb
@@ -53,6 +53,7 @@ module Que
           FROM que_jobs AS j
           WHERE queue = $1::text
           AND job_id >= $2
+          AND run_at >= $3::timestamptz
           AND run_at <= now()
           AND retryable = true
           ORDER BY priority, run_at, job_id
@@ -65,6 +66,7 @@ module Que
               SELECT j
               FROM que_jobs AS j
               WHERE queue = $1::text
+              AND run_at >= $3::timestamptz
               AND run_at <= now()
               AND retryable = true
               AND (priority, run_at, job_id) > (jobs.priority, jobs.run_at, jobs.job_id)
@@ -183,6 +185,7 @@ module Que
             AND run_at <= now()
             AND retryable = true
             AND job_id >= $2
+            AND run_at >= $3::timestamptz
             ORDER BY priority, run_at, job_id
             FOR UPDATE SKIP LOCKED
             LIMIT 1

From 3df1697470fa186dd606a8a6d01ef4f04971b0b6 Mon Sep 17 00:00:00 2001
From: BBrannick <bbrannick@gocardless.com>
Date: Thu, 18 Jun 2026 16:11:24 +0100
Subject: [PATCH 02/11] Add run_at cursor to Locker as an opt-in parallel to
 the job_id cursor

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
---
 lib/que/locker.rb           |  9 ++++--
 spec/lib/que/locker_spec.rb | 56 +++++++++++++++++++++++++++++++++++--
 2 files changed, 60 insertions(+), 5 deletions(-)

diff --git a/lib/que/locker.rb b/lib/que/locker.rb
index f93968ac..e23bf070 100644
--- a/lib/que/locker.rb
+++ b/lib/que/locker.rb
@@ -53,10 +53,12 @@ class Locker
       ),
     ].freeze
 
-    def initialize(queue:, cursor_expiry:, window: nil, budget: nil, secondary_queues: [])
+    def initialize(queue:, cursor_expiry:, window: nil, budget: nil, secondary_queues: [], run_at_cursor: false)
       @queue = queue
       @cursor_expiry = cursor_expiry
+      @run_at_cursor = run_at_cursor
       @queue_cursors = {}
+      @queue_run_at_cursors = {}
       @queue_expires_at = {}
       @secondary_queues = secondary_queues
       @consolidated_queues = Array.wrap(queue).concat(secondary_queues)
@@ -116,6 +118,7 @@ def with_locked_job
       return if job && !exists?(job)
 
       @queue_cursors[job[:queue]] = job[:job_id] if job
+      @queue_run_at_cursors[job[:queue]] = job[:run_at] if job && @run_at_cursor
 
       yield job
     ensure
@@ -149,7 +152,8 @@ def exists?(job)
     end
 
     def lock_job_query(queue, cursor)
-      Que.execute(:lock_job, [queue, cursor]).first
+      run_at = @queue_run_at_cursors.fetch(queue, '-infinity')
+      Que.execute(:lock_job, [queue, cursor, run_at]).first
     end
 
     def handle_expired_cursors!
@@ -161,6 +165,7 @@ def handle_expired_cursors!
 
     def reset_cursor_for!(queue)
       @queue_cursors[queue] = 0
+      @queue_run_at_cursors[queue] = '-infinity'
       @queue_expires_at[queue] = monotonic_now + @cursor_expiry
     end
 
diff --git a/spec/lib/que/locker_spec.rb b/spec/lib/que/locker_spec.rb
index 5e70a8cd..b9aeb5cc 100644
--- a/spec/lib/que/locker_spec.rb
+++ b/spec/lib/que/locker_spec.rb
@@ -43,13 +43,13 @@ def expect_to_work(job)
     end
 
     # Our tests are very concerned with which cursor we use and when
-    def expect_to_lock_with(cursor:)
-      expect(Que).to receive(:execute).with(:lock_job, [queue, cursor])
+    def expect_to_lock_with(cursor:, run_at_cursor: '-infinity')
+      expect(Que).to receive(:execute).with(:lock_job, [queue, cursor, run_at_cursor])
     end
 
     context "with no jobs to lock" do
       it "scans entire table and calls block with nil job" do
-        expect(Que).to receive(:execute).with(:lock_job, [queue, 0])
+        expect(Que).to receive(:execute).with(:lock_job, [queue, 0, '-infinity'])
 
         with_locked_job do |job|
           expect(job).to be_nil
@@ -128,5 +128,55 @@ def expect_to_lock_with(cursor:)
       # rubocop:enable RSpec/SubjectStub
       # rubocop:enable RSpec/InstanceVariable
     end
+
+    context "with run_at_cursor enabled" do
+      subject(:locker) do
+        described_class.new(queue: queue, cursor_expiry: 60, run_at_cursor: true)
+      end
+
+      let!(:job_1) { FakeJob.enqueue(1, queue: queue, priority: 1).attrs }
+      let!(:job_2) { FakeJob.enqueue(2, queue: queue, priority: 2).attrs }
+
+      before do
+        allow(Process).to receive(:clock_gettime).and_return(0)
+        locker.instance_variable_get(:@queue_expires_at)[queue] = 60
+      end
+
+      it "advances the run_at cursor to the previous job's run_at after locking" do
+        expect_to_lock_with(cursor: 0, run_at_cursor: '-infinity')
+        expect_to_work(job_1)
+
+        expect_to_lock_with(cursor: job_1[:job_id], run_at_cursor: job_1[:run_at])
+        with_locked_job { |_job| }
+      end
+
+      it "resets both cursors when the expiry elapses" do
+        expect_to_lock_with(cursor: 0, run_at_cursor: '-infinity')
+        expect_to_work(job_1)
+
+        allow(Process).to receive(:clock_gettime).and_return(61)
+
+        expect_to_lock_with(cursor: 0, run_at_cursor: '-infinity')
+        with_locked_job { |_job| }
+      end
+    end
+
+    context "with run_at_cursor disabled (default)" do
+      let!(:job_1) { FakeJob.enqueue(1, queue: queue, priority: 1).attrs }
+      let(:cursor_expiry) { 60 }
+
+      before do
+        allow(Process).to receive(:clock_gettime).and_return(0)
+        locker.instance_variable_get(:@queue_expires_at)[queue] = 60
+      end
+
+      it "always passes -infinity as the run_at cursor regardless of jobs worked" do
+        expect_to_lock_with(cursor: 0, run_at_cursor: '-infinity')
+        expect_to_work(job_1)
+
+        expect_to_lock_with(cursor: job_1[:job_id], run_at_cursor: '-infinity')
+        with_locked_job { |_job| }
+      end
+    end
   end
 end

From 17e96cbcc0c6bc712c75492ced787ba2709b90a6 Mon Sep 17 00:00:00 2001
From: BBrannick <bbrannick@gocardless.com>
Date: Fri, 19 Jun 2026 11:35:20 +0100
Subject: [PATCH 03/11] Update ActiveRecordWithLock adapter to pass run_at
 cursor to find_job_to_lock

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
---
 lib/que/adapters/active_record_with_lock.rb           |  8 ++++----
 spec/lib/que/adapters/active_record_with_lock_spec.rb | 11 ++++++++++-
 2 files changed, 14 insertions(+), 5 deletions(-)

diff --git a/lib/que/adapters/active_record_with_lock.rb b/lib/que/adapters/active_record_with_lock.rb
index f23ad21c..c44cfdea 100644
--- a/lib/que/adapters/active_record_with_lock.rb
+++ b/lib/que/adapters/active_record_with_lock.rb
@@ -36,8 +36,8 @@ def checkout_lock_database_connection(&block)
       def execute(command, params = [])
         case command
         when :lock_job
-          queue, cursor = params
-          lock_job_with_lock_database(queue, cursor)
+          queue, cursor, run_at_cursor = params
+          lock_job_with_lock_database(queue, cursor, run_at_cursor)
         when :unlock_job
           job_id = params[0]
           unlock_job(job_id)
@@ -48,11 +48,11 @@ def execute(command, params = [])
 
       # This method continues looping through the que_jobs table until it either
       # locks a job successfully or determines that there are no jobs to process.
-      def lock_job_with_lock_database(queue, cursor)
+      def lock_job_with_lock_database(queue, cursor, run_at_cursor = '-infinity')
         loop do
           observe(duration_metric: FindJobSecondsTotal, labels: { queue: queue }) do
             Que.transaction do
-              job_to_lock = Que.execute(:find_job_to_lock, [queue, cursor])
+              job_to_lock = Que.execute(:find_job_to_lock, [queue, cursor, run_at_cursor])
               return job_to_lock if job_to_lock.empty?
 
               cursor = job_to_lock.first["job_id"]
diff --git a/spec/lib/que/adapters/active_record_with_lock_spec.rb b/spec/lib/que/adapters/active_record_with_lock_spec.rb
index a2ec6695..e7d68f8a 100644
--- a/spec/lib/que/adapters/active_record_with_lock_spec.rb
+++ b/spec/lib/que/adapters/active_record_with_lock_spec.rb
@@ -30,7 +30,7 @@
   end
 
   describe ".lock_job_with_lock_database" do
-    subject(:lock_job) { adapter.lock_job_with_lock_database("default", 0) }
+    subject(:lock_job) { adapter.lock_job_with_lock_database("default", 0, '-infinity') }
 
     context "with no jobs enqueued" do
       it "exists the loop and sets correct metric values" do
@@ -40,5 +40,14 @@
         expect(described_class::FindJobHitTotal.values[{ :queue => "default", :job_hit => "true" }]).to eq(0.0)
       end
     end
+
+    context "when passing run_at_cursor" do
+      it "passes run_at_cursor through to find_job_to_lock" do
+        run_at_cursor = '2024-01-01 00:00:00'
+        allow(Que).to receive(:execute).and_call_original
+        expect(Que).to receive(:execute).with(:find_job_to_lock, ["default", 0, run_at_cursor]).and_return([])
+        adapter.lock_job_with_lock_database("default", 0, run_at_cursor)
+      end
+    end
   end
 end

From dbcb942b5aae2a1f91894d8e2b87613709782e83 Mon Sep 17 00:00:00 2001
From: BBrannick <bbrannick@gocardless.com>
Date: Fri, 19 Jun 2026 11:40:35 +0100
Subject: [PATCH 04/11] Add --run-at-cursor flag to Que CLI and Worker, wired
 to Locker

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
---
 bin/que                     |  6 ++++++
 lib/que/worker.rb           |  4 +++-
 spec/lib/que/worker_spec.rb | 14 +++++++-------
 3 files changed, 16 insertions(+), 8 deletions(-)

diff --git a/bin/que b/bin/que
index c5a69f9c..eef97f98 100755
--- a/bin/que
+++ b/bin/que
@@ -47,6 +47,10 @@ OptionParser.new do |opts|
     options.lock_budget = lock_budget
   end
 
+  opts.on("--run-at-cursor", "Enable run_at cursor to skip jobs scheduled before the previous job's run_at (default: false)") do
+    options.run_at_cursor = true
+  end
+
   opts.on("-c", "--worker-count [COUNT]", Integer, "Start this many threaded workers") do |worker_count|
     options.worker_count = worker_count
   end
@@ -100,6 +104,7 @@ log_level        = options.log_level     || ENV["QUE_LOG_LEVEL"]           || "i
 queue_name       = options.queue_name    || ENV["QUE_QUEUE"]               || Que::Worker::DEFAULT_QUEUE
 wake_interval    = options.wake_interval || ENV["QUE_WAKE_INTERVAL"]&.to_f || Que::Worker::DEFAULT_WAKE_INTERVAL
 cursor_expiry    = options.cursor_expiry || wake_interval
+run_at_cursor    = options.run_at_cursor || false
 worker_count     = options.worker_count  || 1
 timeout          = options.timeout
 secondary_queues = options.secondary_queues || []
@@ -128,6 +133,7 @@ worker_group = Que::WorkerGroup.start(
   lock_window: options.lock_window,
   lock_budget: options.lock_budget,
   secondary_queues: secondary_queues,
+  run_at_cursor: run_at_cursor,
 )
 
 if options.metrics_port
diff --git a/lib/que/worker.rb b/lib/que/worker.rb
index ba294e06..8c777167 100644
--- a/lib/que/worker.rb
+++ b/lib/que/worker.rb
@@ -121,7 +121,8 @@ def initialize(
       lock_cursor_expiry: DEFAULT_WAKE_INTERVAL,
       lock_window: nil,
       lock_budget: nil,
-      secondary_queues: []
+      secondary_queues: [],
+      run_at_cursor: false
     )
       @queue = queue
       @wake_interval = wake_interval
@@ -136,6 +137,7 @@ def initialize(
         window: lock_window,
         budget: lock_budget,
         secondary_queues: secondary_queues,
+        run_at_cursor: run_at_cursor,
       )
     end
 
diff --git a/spec/lib/que/worker_spec.rb b/spec/lib/que/worker_spec.rb
index d2f0e152..1559f233 100644
--- a/spec/lib/que/worker_spec.rb
+++ b/spec/lib/que/worker_spec.rb
@@ -211,7 +211,7 @@
           FakeJob.enqueue(1)
 
           expect(Que).
-            to receive(:execute).with(:lock_job, ["default", 0]).and_raise(PG::Error)
+            to receive(:execute).with(:lock_job, ["default", 0, "-infinity"]).and_raise(PG::Error)
           expect(work).to eq(:postgres_error)
         end
       end
@@ -219,7 +219,7 @@
       context "when postgres raises a bad connection error while processing a job" do
         before do
           allow(Que).to receive(:execute).
-            with(:lock_job, ["default", 0]).
+            with(:lock_job, ["default", 0, "-infinity"]).
             and_raise(PG::ConnectionBad)
 
           # Ensure we don't have any currently leased connections, since in a thread
@@ -246,7 +246,7 @@
           FakeJob.enqueue(1)
 
           expect(Que).
-            to receive(:execute).with(:lock_job, ["default", 0]).
+            to receive(:execute).with(:lock_job, ["default", 0, "-infinity"]).
             and_raise(ActiveRecord::ConnectionTimeoutError)
           expect(work).to eq(:postgres_error)
         end
@@ -257,7 +257,7 @@
           FakeJob.enqueue(1)
 
           expect(Que).
-            to receive(:execute).with(:lock_job, ["default", 0]).
+            to receive(:execute).with(:lock_job, ["default", 0, "-infinity"]).
             and_raise(ActiveRecord::ConnectionNotEstablished)
           expect(work).to eq(:postgres_error)
         end
@@ -286,18 +286,18 @@
     it "is false after a postgres error" do
       FakeJob.enqueue(1)
       expect(Que).
-        to receive(:execute).with(:lock_job, ["default", 0]).and_raise(PG::Error)
+        to receive(:execute).with(:lock_job, ["default", 0, "-infinity"]).and_raise(PG::Error)
       worker.work
       expect(worker).to_not be_healthy
     end
 
     it "recovers once work succeeds after a postgres error" do
       expect(Que).
-        to receive(:execute).with(:lock_job, ["default", 0]).and_raise(PG::Error)
+        to receive(:execute).with(:lock_job, ["default", 0, "-infinity"]).and_raise(PG::Error)
       worker.work
       expect(worker).to_not be_healthy
 
-      allow(Que).to receive(:execute).with(:lock_job, ["default", 0]).and_return([])
+      allow(Que).to receive(:execute).with(:lock_job, ["default", 0, "-infinity"]).and_return([])
       worker.work # no job found, returns :job_not_found
       expect(worker).to be_healthy
     end

From cf13c15a5fc97662e9a64cabb2b981f8807e58ec Mon Sep 17 00:00:00 2001
From: BBrannick <bbrannick@gocardless.com>
Date: Mon, 22 Jun 2026 09:51:54 +0100
Subject: [PATCH 05/11] Fix minor review findings: remove unused job_2, clarify
 CLI description

- Remove unused job_2 let binding in locker_spec.rb "with run_at_cursor enabled" context
- Update CLI description to clarify "due before" instead of "scheduled before"

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
---
 bin/que                     | 2 +-
 spec/lib/que/locker_spec.rb | 1 -
 2 files changed, 1 insertion(+), 2 deletions(-)

diff --git a/bin/que b/bin/que
index eef97f98..307076ac 100755
--- a/bin/que
+++ b/bin/que
@@ -47,7 +47,7 @@ OptionParser.new do |opts|
     options.lock_budget = lock_budget
   end
 
-  opts.on("--run-at-cursor", "Enable run_at cursor to skip jobs scheduled before the previous job's run_at (default: false)") do
+  opts.on("--run-at-cursor", "Enable run_at cursor to skip jobs due before the previous job's run_at (default: false)") do
     options.run_at_cursor = true
   end
 
diff --git a/spec/lib/que/locker_spec.rb b/spec/lib/que/locker_spec.rb
index b9aeb5cc..f29d32d8 100644
--- a/spec/lib/que/locker_spec.rb
+++ b/spec/lib/que/locker_spec.rb
@@ -135,7 +135,6 @@ def expect_to_lock_with(cursor:, run_at_cursor: '-infinity')
       end
 
       let!(:job_1) { FakeJob.enqueue(1, queue: queue, priority: 1).attrs }
-      let!(:job_2) { FakeJob.enqueue(2, queue: queue, priority: 2).attrs }
 
       before do
         allow(Process).to receive(:clock_gettime).and_return(0)

From 9dc550a21daad5afd54d7ecfcbf4e1e9553bfc3e Mon Sep 17 00:00:00 2001
From: BBrannick <bbrannick@gocardless.com>
Date: Mon, 22 Jun 2026 12:04:48 +0100
Subject: [PATCH 06/11] Thread run_at_cursor explicitly through
 lock_job_in/execute_lock_job_in/lock_job_query

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
---
 lib/que/locker.rb | 18 +++++++++---------
 1 file changed, 9 insertions(+), 9 deletions(-)

diff --git a/lib/que/locker.rb b/lib/que/locker.rb
index e23bf070..a009bc64 100644
--- a/lib/que/locker.rb
+++ b/lib/que/locker.rb
@@ -80,7 +80,8 @@ def with_locked_job
 
       job = @consolidated_queues.lazy.filter_map do |queue|
         cursor = @queue_cursors.fetch(queue, 0)
-        found_job = lock_job_in(queue, cursor)
+        run_at_cursor = @queue_run_at_cursors.fetch(queue, '-infinity')
+        found_job = lock_job_in(queue, cursor, run_at_cursor)
 
         # Because we were using a cursor when we tried to lock this job, if we fail to
         # find a job it is not necessarily the case that there aren't jobs in the
@@ -96,7 +97,7 @@ def with_locked_job
         # do the appropriate book keeping.
         if found_job.nil? && cursor != 0
           reset_cursor_for!(queue)
-          found_job = lock_job_in(queue, 0)
+          found_job = lock_job_in(queue, 0, '-infinity')
         end
 
         found_job
@@ -131,17 +132,17 @@ def with_locked_job
 
     private
 
-    def lock_job_in(queue, cursor)
+    def lock_job_in(queue, cursor, run_at_cursor)
       observe(nil, ThrottleSecondsTotal, worked_queue: queue) { @leaky_bucket.refill }
-      @leaky_bucket.observe { execute_lock_job_in(queue, cursor) }
+      @leaky_bucket.observe { execute_lock_job_in(queue, cursor, run_at_cursor) }
     end
 
-    def execute_lock_job_in(queue, cursor)
+    def execute_lock_job_in(queue, cursor, run_at_cursor)
       strategy = cursor.zero? ? "full" : "cursor"
       observe(AcquireTotal, AcquireSecondsTotal,
               worked_queue: queue,
               strategy: strategy) do
-        lock_job_query(queue, cursor)
+        lock_job_query(queue, cursor, run_at_cursor)
       end
     end
 
@@ -151,9 +152,8 @@ def exists?(job)
       end
     end
 
-    def lock_job_query(queue, cursor)
-      run_at = @queue_run_at_cursors.fetch(queue, '-infinity')
-      Que.execute(:lock_job, [queue, cursor, run_at]).first
+    def lock_job_query(queue, cursor, run_at_cursor)
+      Que.execute(:lock_job, [queue, cursor, run_at_cursor]).first
     end
 
     def handle_expired_cursors!

From b8d06e46941986dd7d77766438cb21020865843b Mon Sep 17 00:00:00 2001
From: BBrannick <bbrannick@gocardless.com>
Date: Mon, 22 Jun 2026 12:13:47 +0100
Subject: [PATCH 07/11] Use RUN_AT_CURSOR_RESET constant for run_at cursor
 sentinel value

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
---
 lib/que/adapters/active_record_with_lock.rb | 2 +-
 lib/que/locker.rb                           | 8 +++++---
 2 files changed, 6 insertions(+), 4 deletions(-)

diff --git a/lib/que/adapters/active_record_with_lock.rb b/lib/que/adapters/active_record_with_lock.rb
index c44cfdea..8a4afbd0 100644
--- a/lib/que/adapters/active_record_with_lock.rb
+++ b/lib/que/adapters/active_record_with_lock.rb
@@ -48,7 +48,7 @@ def execute(command, params = [])
 
       # This method continues looping through the que_jobs table until it either
       # locks a job successfully or determines that there are no jobs to process.
-      def lock_job_with_lock_database(queue, cursor, run_at_cursor = '-infinity')
+      def lock_job_with_lock_database(queue, cursor, run_at_cursor = Que::Locker::RUN_AT_CURSOR_RESET)
         loop do
           observe(duration_metric: FindJobSecondsTotal, labels: { queue: queue }) do
             Que.transaction do
diff --git a/lib/que/locker.rb b/lib/que/locker.rb
index a009bc64..374bce46 100644
--- a/lib/que/locker.rb
+++ b/lib/que/locker.rb
@@ -15,6 +15,8 @@ module Que
   # For more information, see the 'Predicate Specificity' chapter of:
   # https://brandur.org/postgres-queues
   class Locker
+    RUN_AT_CURSOR_RESET = RUN_AT_CURSOR_RESET
+
     METRICS = [
       ExistsTotal = Prometheus::Client::Counter.new(
         :que_locker_exists_total,
@@ -80,7 +82,7 @@ def with_locked_job
 
       job = @consolidated_queues.lazy.filter_map do |queue|
         cursor = @queue_cursors.fetch(queue, 0)
-        run_at_cursor = @queue_run_at_cursors.fetch(queue, '-infinity')
+        run_at_cursor = @queue_run_at_cursors.fetch(queue, RUN_AT_CURSOR_RESET)
         found_job = lock_job_in(queue, cursor, run_at_cursor)
 
         # Because we were using a cursor when we tried to lock this job, if we fail to
@@ -97,7 +99,7 @@ def with_locked_job
         # do the appropriate book keeping.
         if found_job.nil? && cursor != 0
           reset_cursor_for!(queue)
-          found_job = lock_job_in(queue, 0, '-infinity')
+          found_job = lock_job_in(queue, 0, RUN_AT_CURSOR_RESET)
         end
 
         found_job
@@ -165,7 +167,7 @@ def handle_expired_cursors!
 
     def reset_cursor_for!(queue)
       @queue_cursors[queue] = 0
-      @queue_run_at_cursors[queue] = '-infinity'
+      @queue_run_at_cursors[queue] = RUN_AT_CURSOR_RESET
       @queue_expires_at[queue] = monotonic_now + @cursor_expiry
     end
 

From 7c2f2bc5b232a8fbe60d2609f775a2ed875f479d Mon Sep 17 00:00:00 2001
From: BBrannick <bbrannick@gocardless.com>
Date: Mon, 22 Jun 2026 12:35:14 +0100
Subject: [PATCH 08/11] Diffocop fixes

---
 .../que/adapters/active_record_with_lock_spec.rb   |  4 ++--
 spec/lib/que/locker_spec.rb                        | 14 +++++++-------
 2 files changed, 9 insertions(+), 9 deletions(-)

diff --git a/spec/lib/que/adapters/active_record_with_lock_spec.rb b/spec/lib/que/adapters/active_record_with_lock_spec.rb
index e7d68f8a..527a0a30 100644
--- a/spec/lib/que/adapters/active_record_with_lock_spec.rb
+++ b/spec/lib/que/adapters/active_record_with_lock_spec.rb
@@ -30,7 +30,7 @@
   end
 
   describe ".lock_job_with_lock_database" do
-    subject(:lock_job) { adapter.lock_job_with_lock_database("default", 0, '-infinity') }
+    subject(:lock_job) { adapter.lock_job_with_lock_database("default", 0, "-infinity") }
 
     context "with no jobs enqueued" do
       it "exists the loop and sets correct metric values" do
@@ -43,7 +43,7 @@
 
     context "when passing run_at_cursor" do
       it "passes run_at_cursor through to find_job_to_lock" do
-        run_at_cursor = '2024-01-01 00:00:00'
+        run_at_cursor = "2024-01-01 00:00:00"
         allow(Que).to receive(:execute).and_call_original
         expect(Que).to receive(:execute).with(:find_job_to_lock, ["default", 0, run_at_cursor]).and_return([])
         adapter.lock_job_with_lock_database("default", 0, run_at_cursor)
diff --git a/spec/lib/que/locker_spec.rb b/spec/lib/que/locker_spec.rb
index f29d32d8..be6d02ed 100644
--- a/spec/lib/que/locker_spec.rb
+++ b/spec/lib/que/locker_spec.rb
@@ -43,13 +43,13 @@ def expect_to_work(job)
     end
 
     # Our tests are very concerned with which cursor we use and when
-    def expect_to_lock_with(cursor:, run_at_cursor: '-infinity')
+    def expect_to_lock_with(cursor:, run_at_cursor: "-infinity")
       expect(Que).to receive(:execute).with(:lock_job, [queue, cursor, run_at_cursor])
     end
 
     context "with no jobs to lock" do
       it "scans entire table and calls block with nil job" do
-        expect(Que).to receive(:execute).with(:lock_job, [queue, 0, '-infinity'])
+        expect(Que).to receive(:execute).with(:lock_job, [queue, 0, "-infinity"])
 
         with_locked_job do |job|
           expect(job).to be_nil
@@ -142,7 +142,7 @@ def expect_to_lock_with(cursor:, run_at_cursor: '-infinity')
       end
 
       it "advances the run_at cursor to the previous job's run_at after locking" do
-        expect_to_lock_with(cursor: 0, run_at_cursor: '-infinity')
+        expect_to_lock_with(cursor: 0, run_at_cursor: "-infinity")
         expect_to_work(job_1)
 
         expect_to_lock_with(cursor: job_1[:job_id], run_at_cursor: job_1[:run_at])
@@ -150,12 +150,12 @@ def expect_to_lock_with(cursor:, run_at_cursor: '-infinity')
       end
 
       it "resets both cursors when the expiry elapses" do
-        expect_to_lock_with(cursor: 0, run_at_cursor: '-infinity')
+        expect_to_lock_with(cursor: 0, run_at_cursor: "-infinity")
         expect_to_work(job_1)
 
         allow(Process).to receive(:clock_gettime).and_return(61)
 
-        expect_to_lock_with(cursor: 0, run_at_cursor: '-infinity')
+        expect_to_lock_with(cursor: 0, run_at_cursor: "-infinity")
         with_locked_job { |_job| }
       end
     end
@@ -170,10 +170,10 @@ def expect_to_lock_with(cursor:, run_at_cursor: '-infinity')
       end
 
       it "always passes -infinity as the run_at cursor regardless of jobs worked" do
-        expect_to_lock_with(cursor: 0, run_at_cursor: '-infinity')
+        expect_to_lock_with(cursor: 0, run_at_cursor: "-infinity")
         expect_to_work(job_1)
 
-        expect_to_lock_with(cursor: job_1[:job_id], run_at_cursor: '-infinity')
+        expect_to_lock_with(cursor: job_1[:job_id], run_at_cursor: "-infinity")
         with_locked_job { |_job| }
       end
     end

From 379b044d6c4408ea6f048ff1588a87ef1ac3f017 Mon Sep 17 00:00:00 2001
From: BBrannick <bbrannick@gocardless.com>
Date: Mon, 22 Jun 2026 12:39:46 +0100
Subject: [PATCH 09/11] Fix RUN_AT_CURSOR_RESET constant self-assignment

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
---
 lib/que/locker.rb | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/lib/que/locker.rb b/lib/que/locker.rb
index 374bce46..f475d65a 100644
--- a/lib/que/locker.rb
+++ b/lib/que/locker.rb
@@ -15,7 +15,7 @@ module Que
   # For more information, see the 'Predicate Specificity' chapter of:
   # https://brandur.org/postgres-queues
   class Locker
-    RUN_AT_CURSOR_RESET = RUN_AT_CURSOR_RESET
+    RUN_AT_CURSOR_RESET = "-infinity"
 
     METRICS = [
       ExistsTotal = Prometheus::Client::Counter.new(

From 033e384d38f61d34065f40b55c7eb1f5660d3acb Mon Sep 17 00:00:00 2001
From: BBrannick <bbrannick@gocardless.com>
Date: Tue, 23 Jun 2026 16:51:04 +0100
Subject: [PATCH 10/11] Rename run_at_cursor to run_at_lower_bound where it
 represents a timestamp value

run_at_cursor is now only used as the boolean feature flag on Locker/Worker.
The actual timestamp lower bound passed through the locking methods is
renamed run_at_lower_bound to make the distinction clear.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
---
 lib/que/adapters/active_record_with_lock.rb |  8 ++++----
 lib/que/locker.rb                           | 22 ++++++++++-----------
 spec/lib/que/locker_spec.rb                 | 16 +++++++--------
 3 files changed, 23 insertions(+), 23 deletions(-)

diff --git a/lib/que/adapters/active_record_with_lock.rb b/lib/que/adapters/active_record_with_lock.rb
index 8a4afbd0..a7bfaedf 100644
--- a/lib/que/adapters/active_record_with_lock.rb
+++ b/lib/que/adapters/active_record_with_lock.rb
@@ -36,8 +36,8 @@ def checkout_lock_database_connection(&block)
       def execute(command, params = [])
         case command
         when :lock_job
-          queue, cursor, run_at_cursor = params
-          lock_job_with_lock_database(queue, cursor, run_at_cursor)
+          queue, cursor, run_at_lower_bound = params
+          lock_job_with_lock_database(queue, cursor, run_at_lower_bound)
         when :unlock_job
           job_id = params[0]
           unlock_job(job_id)
@@ -48,11 +48,11 @@ def execute(command, params = [])
 
       # This method continues looping through the que_jobs table until it either
       # locks a job successfully or determines that there are no jobs to process.
-      def lock_job_with_lock_database(queue, cursor, run_at_cursor = Que::Locker::RUN_AT_CURSOR_RESET)
+      def lock_job_with_lock_database(queue, cursor, run_at_lower_bound = Que::Locker::RUN_AT_CURSOR_RESET)
         loop do
           observe(duration_metric: FindJobSecondsTotal, labels: { queue: queue }) do
             Que.transaction do
-              job_to_lock = Que.execute(:find_job_to_lock, [queue, cursor, run_at_cursor])
+              job_to_lock = Que.execute(:find_job_to_lock, [queue, cursor, run_at_lower_bound])
               return job_to_lock if job_to_lock.empty?
 
               cursor = job_to_lock.first["job_id"]
diff --git a/lib/que/locker.rb b/lib/que/locker.rb
index f475d65a..4d4c5afa 100644
--- a/lib/que/locker.rb
+++ b/lib/que/locker.rb
@@ -60,7 +60,7 @@ def initialize(queue:, cursor_expiry:, window: nil, budget: nil, secondary_queue
       @cursor_expiry = cursor_expiry
       @run_at_cursor = run_at_cursor
       @queue_cursors = {}
-      @queue_run_at_cursors = {}
+      @run_at_lower_bounds = {}
       @queue_expires_at = {}
       @secondary_queues = secondary_queues
       @consolidated_queues = Array.wrap(queue).concat(secondary_queues)
@@ -82,8 +82,8 @@ def with_locked_job
 
       job = @consolidated_queues.lazy.filter_map do |queue|
         cursor = @queue_cursors.fetch(queue, 0)
-        run_at_cursor = @queue_run_at_cursors.fetch(queue, RUN_AT_CURSOR_RESET)
-        found_job = lock_job_in(queue, cursor, run_at_cursor)
+        run_at_lower_bound = @run_at_lower_bounds.fetch(queue, RUN_AT_CURSOR_RESET)
+        found_job = lock_job_in(queue, cursor, run_at_lower_bound)
 
         # Because we were using a cursor when we tried to lock this job, if we fail to
         # find a job it is not necessarily the case that there aren't jobs in the
@@ -121,7 +121,7 @@ def with_locked_job
       return if job && !exists?(job)
 
       @queue_cursors[job[:queue]] = job[:job_id] if job
-      @queue_run_at_cursors[job[:queue]] = job[:run_at] if job && @run_at_cursor
+      @run_at_lower_bounds[job[:queue]] = job[:run_at] if job && @run_at_cursor
 
       yield job
     ensure
@@ -134,17 +134,17 @@ def with_locked_job
 
     private
 
-    def lock_job_in(queue, cursor, run_at_cursor)
+    def lock_job_in(queue, cursor, run_at_lower_bound)
       observe(nil, ThrottleSecondsTotal, worked_queue: queue) { @leaky_bucket.refill }
-      @leaky_bucket.observe { execute_lock_job_in(queue, cursor, run_at_cursor) }
+      @leaky_bucket.observe { execute_lock_job_in(queue, cursor, run_at_lower_bound) }
     end
 
-    def execute_lock_job_in(queue, cursor, run_at_cursor)
+    def execute_lock_job_in(queue, cursor, run_at_lower_bound)
       strategy = cursor.zero? ? "full" : "cursor"
       observe(AcquireTotal, AcquireSecondsTotal,
               worked_queue: queue,
               strategy: strategy) do
-        lock_job_query(queue, cursor, run_at_cursor)
+        lock_job_query(queue, cursor, run_at_lower_bound)
       end
     end
 
@@ -154,8 +154,8 @@ def exists?(job)
       end
     end
 
-    def lock_job_query(queue, cursor, run_at_cursor)
-      Que.execute(:lock_job, [queue, cursor, run_at_cursor]).first
+    def lock_job_query(queue, cursor, run_at_lower_bound)
+      Que.execute(:lock_job, [queue, cursor, run_at_lower_bound]).first
     end
 
     def handle_expired_cursors!
@@ -167,7 +167,7 @@ def handle_expired_cursors!
 
     def reset_cursor_for!(queue)
       @queue_cursors[queue] = 0
-      @queue_run_at_cursors[queue] = RUN_AT_CURSOR_RESET
+      @run_at_lower_bounds[queue] = RUN_AT_CURSOR_RESET
       @queue_expires_at[queue] = monotonic_now + @cursor_expiry
     end
 
diff --git a/spec/lib/que/locker_spec.rb b/spec/lib/que/locker_spec.rb
index be6d02ed..a7cc7164 100644
--- a/spec/lib/que/locker_spec.rb
+++ b/spec/lib/que/locker_spec.rb
@@ -43,8 +43,8 @@ def expect_to_work(job)
     end
 
     # Our tests are very concerned with which cursor we use and when
-    def expect_to_lock_with(cursor:, run_at_cursor: "-infinity")
-      expect(Que).to receive(:execute).with(:lock_job, [queue, cursor, run_at_cursor])
+    def expect_to_lock_with(cursor:, run_at_lower_bound: "-infinity")
+      expect(Que).to receive(:execute).with(:lock_job, [queue, cursor, run_at_lower_bound])
     end
 
     context "with no jobs to lock" do
@@ -142,20 +142,20 @@ def expect_to_lock_with(cursor:, run_at_cursor: "-infinity")
       end
 
       it "advances the run_at cursor to the previous job's run_at after locking" do
-        expect_to_lock_with(cursor: 0, run_at_cursor: "-infinity")
+        expect_to_lock_with(cursor: 0, run_at_lower_bound: "-infinity")
         expect_to_work(job_1)
 
-        expect_to_lock_with(cursor: job_1[:job_id], run_at_cursor: job_1[:run_at])
+        expect_to_lock_with(cursor: job_1[:job_id], run_at_lower_bound: job_1[:run_at])
         with_locked_job { |_job| }
       end
 
       it "resets both cursors when the expiry elapses" do
-        expect_to_lock_with(cursor: 0, run_at_cursor: "-infinity")
+        expect_to_lock_with(cursor: 0, run_at_lower_bound: "-infinity")
         expect_to_work(job_1)
 
         allow(Process).to receive(:clock_gettime).and_return(61)
 
-        expect_to_lock_with(cursor: 0, run_at_cursor: "-infinity")
+        expect_to_lock_with(cursor: 0, run_at_lower_bound: "-infinity")
         with_locked_job { |_job| }
       end
     end
@@ -170,10 +170,10 @@ def expect_to_lock_with(cursor:, run_at_cursor: "-infinity")
       end
 
       it "always passes -infinity as the run_at cursor regardless of jobs worked" do
-        expect_to_lock_with(cursor: 0, run_at_cursor: "-infinity")
+        expect_to_lock_with(cursor: 0, run_at_lower_bound: "-infinity")
         expect_to_work(job_1)
 
-        expect_to_lock_with(cursor: job_1[:job_id], run_at_cursor: "-infinity")
+        expect_to_lock_with(cursor: job_1[:job_id], run_at_lower_bound: "-infinity")
         with_locked_job { |_job| }
       end
     end

From c282f547b9e244ffdca44d80b93ce48b5abd9fb5 Mon Sep 17 00:00:00 2001
From: BBrannick <bbrannick@gocardless.com>
Date: Tue, 23 Jun 2026 17:49:53 +0100
Subject: [PATCH 11/11] Use nil as constant default for run_at cursor

---
 lib/que/locker.rb                                  |  2 +-
 lib/que/sql.rb                                     |  6 +++---
 .../que/adapters/active_record_with_lock_spec.rb   |  2 +-
 spec/lib/que/locker_spec.rb                        | 14 +++++++-------
 spec/lib/que/worker_spec.rb                        | 14 +++++++-------
 5 files changed, 19 insertions(+), 19 deletions(-)

diff --git a/lib/que/locker.rb b/lib/que/locker.rb
index 4d4c5afa..65ebc51f 100644
--- a/lib/que/locker.rb
+++ b/lib/que/locker.rb
@@ -15,7 +15,7 @@ module Que
   # For more information, see the 'Predicate Specificity' chapter of:
   # https://brandur.org/postgres-queues
   class Locker
-    RUN_AT_CURSOR_RESET = "-infinity"
+    RUN_AT_CURSOR_RESET = nil
 
     METRICS = [
       ExistsTotal = Prometheus::Client::Counter.new(
diff --git a/lib/que/sql.rb b/lib/que/sql.rb
index 449efcdc..5dc82df2 100644
--- a/lib/que/sql.rb
+++ b/lib/que/sql.rb
@@ -53,7 +53,7 @@ module Que
           FROM que_jobs AS j
           WHERE queue = $1::text
           AND job_id >= $2
-          AND run_at >= $3::timestamptz
+          AND run_at >= COALESCE($3::timestamptz, '-infinity'::timestamptz)
           AND run_at <= now()
           AND retryable = true
           ORDER BY priority, run_at, job_id
@@ -66,7 +66,7 @@ module Que
               SELECT j
               FROM que_jobs AS j
               WHERE queue = $1::text
-              AND run_at >= $3::timestamptz
+              AND run_at >= COALESCE($3::timestamptz, '-infinity'::timestamptz)
               AND run_at <= now()
               AND retryable = true
               AND (priority, run_at, job_id) > (jobs.priority, jobs.run_at, jobs.job_id)
@@ -185,7 +185,7 @@ module Que
             AND run_at <= now()
             AND retryable = true
             AND job_id >= $2
-            AND run_at >= $3::timestamptz
+            AND run_at >= COALESCE($3::timestamptz, '-infinity'::timestamptz)
             ORDER BY priority, run_at, job_id
             FOR UPDATE SKIP LOCKED
             LIMIT 1
diff --git a/spec/lib/que/adapters/active_record_with_lock_spec.rb b/spec/lib/que/adapters/active_record_with_lock_spec.rb
index 527a0a30..6123ac4e 100644
--- a/spec/lib/que/adapters/active_record_with_lock_spec.rb
+++ b/spec/lib/que/adapters/active_record_with_lock_spec.rb
@@ -30,7 +30,7 @@
   end
 
   describe ".lock_job_with_lock_database" do
-    subject(:lock_job) { adapter.lock_job_with_lock_database("default", 0, "-infinity") }
+    subject(:lock_job) { adapter.lock_job_with_lock_database("default", 0) }
 
     context "with no jobs enqueued" do
       it "exists the loop and sets correct metric values" do
diff --git a/spec/lib/que/locker_spec.rb b/spec/lib/que/locker_spec.rb
index a7cc7164..c8b686d1 100644
--- a/spec/lib/que/locker_spec.rb
+++ b/spec/lib/que/locker_spec.rb
@@ -43,13 +43,13 @@ def expect_to_work(job)
     end
 
     # Our tests are very concerned with which cursor we use and when
-    def expect_to_lock_with(cursor:, run_at_lower_bound: "-infinity")
+    def expect_to_lock_with(cursor:, run_at_lower_bound: nil)
       expect(Que).to receive(:execute).with(:lock_job, [queue, cursor, run_at_lower_bound])
     end
 
     context "with no jobs to lock" do
       it "scans entire table and calls block with nil job" do
-        expect(Que).to receive(:execute).with(:lock_job, [queue, 0, "-infinity"])
+        expect(Que).to receive(:execute).with(:lock_job, [queue, 0, nil])
 
         with_locked_job do |job|
           expect(job).to be_nil
@@ -142,7 +142,7 @@ def expect_to_lock_with(cursor:, run_at_lower_bound: "-infinity")
       end
 
       it "advances the run_at cursor to the previous job's run_at after locking" do
-        expect_to_lock_with(cursor: 0, run_at_lower_bound: "-infinity")
+        expect_to_lock_with(cursor: 0, run_at_lower_bound: nil)
         expect_to_work(job_1)
 
         expect_to_lock_with(cursor: job_1[:job_id], run_at_lower_bound: job_1[:run_at])
@@ -150,12 +150,12 @@ def expect_to_lock_with(cursor:, run_at_lower_bound: "-infinity")
       end
 
       it "resets both cursors when the expiry elapses" do
-        expect_to_lock_with(cursor: 0, run_at_lower_bound: "-infinity")
+        expect_to_lock_with(cursor: 0, run_at_lower_bound: nil)
         expect_to_work(job_1)
 
         allow(Process).to receive(:clock_gettime).and_return(61)
 
-        expect_to_lock_with(cursor: 0, run_at_lower_bound: "-infinity")
+        expect_to_lock_with(cursor: 0, run_at_lower_bound: nil)
         with_locked_job { |_job| }
       end
     end
@@ -170,10 +170,10 @@ def expect_to_lock_with(cursor:, run_at_lower_bound: "-infinity")
       end
 
       it "always passes -infinity as the run_at cursor regardless of jobs worked" do
-        expect_to_lock_with(cursor: 0, run_at_lower_bound: "-infinity")
+        expect_to_lock_with(cursor: 0, run_at_lower_bound: nil)
         expect_to_work(job_1)
 
-        expect_to_lock_with(cursor: job_1[:job_id], run_at_lower_bound: "-infinity")
+        expect_to_lock_with(cursor: job_1[:job_id], run_at_lower_bound: nil)
         with_locked_job { |_job| }
       end
     end
diff --git a/spec/lib/que/worker_spec.rb b/spec/lib/que/worker_spec.rb
index 1559f233..ea603ce7 100644
--- a/spec/lib/que/worker_spec.rb
+++ b/spec/lib/que/worker_spec.rb
@@ -211,7 +211,7 @@
           FakeJob.enqueue(1)
 
           expect(Que).
-            to receive(:execute).with(:lock_job, ["default", 0, "-infinity"]).and_raise(PG::Error)
+            to receive(:execute).with(:lock_job, ["default", 0, nil]).and_raise(PG::Error)
           expect(work).to eq(:postgres_error)
         end
       end
@@ -219,7 +219,7 @@
       context "when postgres raises a bad connection error while processing a job" do
         before do
           allow(Que).to receive(:execute).
-            with(:lock_job, ["default", 0, "-infinity"]).
+            with(:lock_job, ["default", 0, nil]).
             and_raise(PG::ConnectionBad)
 
           # Ensure we don't have any currently leased connections, since in a thread
@@ -246,7 +246,7 @@
           FakeJob.enqueue(1)
 
           expect(Que).
-            to receive(:execute).with(:lock_job, ["default", 0, "-infinity"]).
+            to receive(:execute).with(:lock_job, ["default", 0, nil]).
             and_raise(ActiveRecord::ConnectionTimeoutError)
           expect(work).to eq(:postgres_error)
         end
@@ -257,7 +257,7 @@
           FakeJob.enqueue(1)
 
           expect(Que).
-            to receive(:execute).with(:lock_job, ["default", 0, "-infinity"]).
+            to receive(:execute).with(:lock_job, ["default", 0, nil]).
             and_raise(ActiveRecord::ConnectionNotEstablished)
           expect(work).to eq(:postgres_error)
         end
@@ -286,18 +286,18 @@
     it "is false after a postgres error" do
       FakeJob.enqueue(1)
       expect(Que).
-        to receive(:execute).with(:lock_job, ["default", 0, "-infinity"]).and_raise(PG::Error)
+        to receive(:execute).with(:lock_job, ["default", 0, nil]).and_raise(PG::Error)
       worker.work
       expect(worker).to_not be_healthy
     end
 
     it "recovers once work succeeds after a postgres error" do
       expect(Que).
-        to receive(:execute).with(:lock_job, ["default", 0, "-infinity"]).and_raise(PG::Error)
+        to receive(:execute).with(:lock_job, ["default", 0, nil]).and_raise(PG::Error)
       worker.work
       expect(worker).to_not be_healthy
 
-      allow(Que).to receive(:execute).with(:lock_job, ["default", 0, "-infinity"]).and_return([])
+      allow(Que).to receive(:execute).with(:lock_job, ["default", 0, nil]).and_return([])
       worker.work # no job found, returns :job_not_found
       expect(worker).to be_healthy
     end