Use nil as constant default for run_at cursor

Author: BBrannick

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(

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

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

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,20 +142,20 @@ 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])
         with_locked_job { |_job| }
       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

spec/lib/que/worker_spec.rb

@@ -211,15 +211,15 @@
           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
 
       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