diff --git a/bin/que b/bin/que index c5a69f9c..307076ac 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 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 diff --git a/lib/que/adapters/active_record_with_lock.rb b/lib/que/adapters/active_record_with_lock.rb index f23ad21c..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 = 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"] diff --git a/lib/que/locker.rb b/lib/que/locker.rb index f93968ac..65ebc51f 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 = 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 diff --git a/lib/que/sql.rb b/lib/que/sql.rb index 0de73b6c..5dc82df2 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 >= 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 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/adapters/active_record_with_lock_spec.rb b/spec/lib/que/adapters/active_record_with_lock_spec.rb index a2ec6695..6123ac4e 100644 --- a/spec/lib/que/adapters/active_record_with_lock_spec.rb +++ b/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 diff --git a/spec/lib/que/locker_spec.rb b/spec/lib/que/locker_spec.rb index 5e70a8cd..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:) - 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 diff --git a/spec/lib/que/worker_spec.rb b/spec/lib/que/worker_spec.rb index d2f0e152..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]).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]). + 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