Add optional cursor for run_at time on jobs

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 due 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

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_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)
+      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])
+              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

@@ -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 = nil
+
     METRICS = [
       ExistsTotal = Prometheus::Client::Counter.new(
         :que_locker_exists_total,
@@ -53,10 +55,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 = {}
+      @run_at_lower_bounds = {}
       @queue_expires_at = {}
       @secondary_queues = secondary_queues
       @consolidated_queues = Array.wrap(queue).concat(secondary_queues)
@@ -78,7 +82,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_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
@@ -94,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)
+          found_job = lock_job_in(queue, 0, RUN_AT_CURSOR_RESET)
         end
 
         found_job
@@ -116,6 +121,7 @@ def with_locked_job
       return if job && !exists?(job)
 
       @queue_cursors[job[:queue]] = job[:job_id] if job
+      @run_at_lower_bounds[job[:queue]] = job[:run_at] if job && @run_at_cursor
 
       yield job
     ensure
@@ -128,17 +134,17 @@ def with_locked_job
 
     private
 
-    def lock_job_in(queue, 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) }
+      @leaky_bucket.observe { execute_lock_job_in(queue, cursor, run_at_lower_bound) }
     end
 
-    def execute_lock_job_in(queue, 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)
+        lock_job_query(queue, cursor, run_at_lower_bound)
       end
     end
 
@@ -148,8 +154,8 @@ def exists?(job)
       end
     end
 
-    def lock_job_query(queue, cursor)
-      Que.execute(:lock_job, [queue, 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!
@@ -161,6 +167,7 @@ def handle_expired_cursors!
 
     def reset_cursor_for!(queue)
       @queue_cursors[queue] = 0
+      @run_at_lower_bounds[queue] = RUN_AT_CURSOR_RESET
       @queue_expires_at[queue] = monotonic_now + @cursor_expiry
     end
 

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 >= COALESCE($3::timestamptz, '-infinity'::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 >= 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)
@@ -183,6 +185,7 @@ module Que
             AND run_at <= now()
             AND retryable = true
             AND job_id >= $2
+            AND run_at >= COALESCE($3::timestamptz, '-infinity'::timestamptz)
             ORDER BY priority, run_at, job_id
             FOR UPDATE SKIP LOCKED
             LIMIT 1

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
 

spec/lib/que/adapters/active_record_with_lock_spec.rb

@@ -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

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_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])
+        expect(Que).to receive(:execute).with(:lock_job, [queue, 0, nil])
 
         with_locked_job do |job|
           expect(job).to be_nil
@@ -128,5 +128,54 @@ 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 }
+
+      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_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: 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: nil)
+        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_lower_bound: nil)
+        expect_to_work(job_1)
+
+        expect_to_lock_with(cursor: job_1[:job_id], run_at_lower_bound: nil)
+        with_locked_job { |_job| }
+      end
+    end
   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]).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]).
+            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]).
+            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]).
+            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]).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]).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]).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