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>

Author: BBrannick

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"]

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
 

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
@@ -131,7 +131,7 @@ def expect_to_lock_with(cursor:, run_at_cursor: "-infinity")
 
     context "with run_at_cursor enabled" do
       subject(:locker) do
-        described_class.new(queue: queue, cursor_expiry: 60, run_at_cursor: true)
+        described_class.new(queue: queue, cursor_expiry: 60, run_at_lower_bound: true)
       end
 
       let!(:job_1) { FakeJob.enqueue(1, queue: queue, priority: 1).attrs }
@@ -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