diff --git a/UPGRADING.md b/UPGRADING.md index 2eecfd0c..74672182 100644 --- a/UPGRADING.md +++ b/UPGRADING.md @@ -1,3 +1,10 @@ +# Upgrading to version 1.7.2 +Jobs with [concurrency controls](README.md#concurrency-controls) now remember which semaphore they got their slot from, in their serialized job data, and give the slot back only to that semaphore. Before, a job that ran past its `duration` gave its slot back to the semaphore a later job had created after the job's own one expired, so one more job than the limit allowed could run at the same time, and keep doing so until the jobs waiting for that key ran out. + +There's nothing to change or migrate. Jobs that got their slot before you upgrade have nothing recorded and give it back as before, so old and new versions can run side by side while you deploy. + +Solid Queue 2.0 will stop giving back slots that weren't recorded. Before upgrading to 2.0, make sure the concurrency-limited jobs that were ready or running when you upgraded to 1.7.2 have finished. Otherwise their slots won't be given back: a slot like that only comes back when its semaphore expires, and every job that takes or gives back a slot for the same key pushes that expiry out by another `duration`, so on a busy key with a limit above 1 it can stay lost for as long as the key stays busy. + # Upgrading to version 1.7.x This version introduces support for grouping jobs into batches, which needs new tables. Fresh installs get them with the base schema; existing installations need to copy the migration that adds them and run it: diff --git a/app/models/solid_queue/blocked_execution.rb b/app/models/solid_queue/blocked_execution.rb index a84660a4..fd085349 100644 --- a/app/models/solid_queue/blocked_execution.rb +++ b/app/models/solid_queue/blocked_execution.rb @@ -51,7 +51,7 @@ def release destroy! payload[:failed] = true - elsif acquire_concurrency_lock + elsif job.acquire_concurrency_lock promote_to_ready destroy! @@ -63,12 +63,7 @@ def release private def set_expires_at - duration = job.job_class ? job.concurrency_duration : SolidQueue.default_concurrency_control_period - self.expires_at = duration.from_now - end - - def acquire_concurrency_lock - Semaphore.wait(job) + self.expires_at = job.concurrency_duration.from_now end def promote_to_ready diff --git a/app/models/solid_queue/job/concurrency_controls.rb b/app/models/solid_queue/job/concurrency_controls.rb index 4acec12a..ccf985c0 100644 --- a/app/models/solid_queue/job/concurrency_controls.rb +++ b/app/models/solid_queue/job/concurrency_controls.rb @@ -5,11 +5,12 @@ class Job module ConcurrencyControls extend ActiveSupport::Concern + # Stored in the job's serialized Active Job data, which Active Job ignores when running the job + CONCURRENCY_LOCK_KEY = "solid_queue_concurrency_lock" + included do has_one :blocked_execution - delegate :concurrency_limit, :concurrency_duration, to: :job_class - before_destroy :unblock_next_blocked_job, if: -> { concurrency_limited? && ready? } end @@ -25,8 +26,39 @@ def unblock_next_blocked_job end end + def acquire_concurrency_lock + return true unless concurrency_limited? + + if job_class.nil? + # A job whose class doesn't load skips its limits. Record that it holds no slot, + # so it doesn't give one back if it runs somewhere its class does load. + record_concurrency_lock(semaphore_id: nil) + elsif semaphore_id = Semaphore.wait(self) + record_concurrency_lock(semaphore_id:, limit: concurrency_limit) + end + end + def concurrency_limited? - concurrency_key.present? && job_class.present? + concurrency_key.present? + end + + # A job that kept its concurrency key after its class dropped limits_concurrency gets a limit of 1 + def concurrency_limit + job_class.concurrency_limit || 1 if concurrency_limited? + end + + def concurrency_duration + job_class ? job_class.concurrency_duration : SolidQueue.default_concurrency_control_period + end + + # The semaphore the job got its concurrency slot from + def acquired_semaphore_id + concurrency_lock&.dig("semaphore_id") + end + + # The concurrency limit the job had when it got its slot + def acquired_concurrency_limit + concurrency_lock&.dig("limit") end def blocked? @@ -38,20 +70,35 @@ def job_class end private - def concurrency_on_conflict - job_class.concurrency_on_conflict.to_s.inquiry + def release_concurrency_lock + holds_concurrency_slot? && Semaphore.signal(self) end - def acquire_concurrency_lock - return true unless concurrency_limited? + def holds_concurrency_slot? + if concurrency_lock_recorded? + acquired_semaphore_id.present? + else + # Jobs dispatched before Solid Queue recorded their semaphore, which need their class to know their limit + concurrency_limited? && job_class.present? + end + end - Semaphore.wait(self) + def concurrency_lock_recorded? + arguments.key?(CONCURRENCY_LOCK_KEY) end - def release_concurrency_lock - return false unless concurrency_limited? + def release_next_blocked_job + BlockedExecution.release_one(concurrency_key) + end + + def record_concurrency_lock(semaphore_id:, limit: nil) + arguments[CONCURRENCY_LOCK_KEY] = { "semaphore_id" => semaphore_id, "limit" => limit } + update_column :arguments, arguments + true + end - Semaphore.signal(self) + def concurrency_lock + arguments[CONCURRENCY_LOCK_KEY] end def handle_concurrency_conflict @@ -62,12 +109,12 @@ def handle_concurrency_conflict end end - def block - BlockedExecution.create_or_find_by!(job: self) + def concurrency_on_conflict + job_class.concurrency_on_conflict.to_s.inquiry end - def release_next_blocked_job - BlockedExecution.release_one(concurrency_key) + def block + BlockedExecution.create_or_find_by!(job: self) end def execution diff --git a/app/models/solid_queue/semaphore.rb b/app/models/solid_queue/semaphore.rb index da2ea4c3..4fa6768d 100644 --- a/app/models/solid_queue/semaphore.rb +++ b/app/models/solid_queue/semaphore.rb @@ -31,9 +31,10 @@ def initialize(job) @job = job end + # Returns the id of the semaphore the job got a slot on, or nil if there was none free def wait if semaphore = Semaphore.lock.find_by(key: key) - semaphore.value > 0 && attempt_decrement + semaphore.id if semaphore.value > 0 && attempt_decrement else attempt_creation end @@ -46,24 +47,46 @@ def signal private attr_accessor :job + def attempt_decrement + Semaphore.available.where(key: key).update_all([ "value = value - 1, expires_at = ?", expires_at ]) > 0 + end + def attempt_creation if Semaphore.create_unique_by(key: key, value: limit - 1, expires_at: expires_at) - true + acquired_semaphore_id else check_limit_or_decrement end end - def check_limit_or_decrement - limit == 1 ? false : attempt_decrement + # A locking read, so it sees the latest committed rows rather than the transaction's + # snapshot, which on MySQL can still include a semaphore that expired and was deleted + def acquired_semaphore_id + Semaphore.lock.where(key: key).pick(:id) end - def attempt_decrement - Semaphore.available.where(key: key).update_all([ "value = value - 1, expires_at = ?", expires_at ]) > 0 + def check_limit_or_decrement + acquired_semaphore_id if limit > 1 && attempt_decrement end def attempt_increment - Semaphore.where(key: key, value: ...limit).update_all([ "value = value + 1, expires_at = ?", expires_at ]) > 0 + semaphore_to_release.where(value: ...limit_to_release).update_all([ "value = value + 1, expires_at = ?", expires_at ]) > 0 + end + + # A job only gives its slot back to the semaphore it got it from. If that one expired + # and was deleted, a semaphore for the same key now belongs to other jobs. + # Jobs dispatched before Solid Queue recorded their semaphore give it back by key. + def semaphore_to_release + if job.acquired_semaphore_id + Semaphore.where(id: job.acquired_semaphore_id) + else + Semaphore.where(key: key) + end + end + + # Recorded when the job got its slot, so it's known even if the job's class no longer loads + def limit_to_release + job.acquired_concurrency_limit || limit end def key @@ -75,7 +98,7 @@ def expires_at end def limit - job.concurrency_limit || 1 + job.concurrency_limit end end end diff --git a/test/integration/concurrency_controls_test.rb b/test/integration/concurrency_controls_test.rb index 147100b3..b6383c2e 100644 --- a/test/integration/concurrency_controls_test.rb +++ b/test/integration/concurrency_controls_test.rb @@ -116,7 +116,7 @@ class ConcurrencyControlsTest < ActiveSupport::TestCase test "rely on dispatcher to unblock blocked executions with an available semaphore" do # Simulate a scenario where we got an available semaphore and some stuck jobs - job = NonOverlappingUpdateResultJob.perform_later(@result, name: "A") + job = SolidQueue::Job.find_by!(active_job_id: NonOverlappingUpdateResultJob.perform_later(@result, name: "A").job_id) wait_for_jobs_to_finish_for(5.seconds) assert_no_unfinished_jobs @@ -148,7 +148,7 @@ class ConcurrencyControlsTest < ActiveSupport::TestCase test "rely on dispatcher to unblock blocked executions with an expired semaphore" do # Simulate a scenario where we got an available semaphore and some stuck jobs - job = NonOverlappingUpdateResultJob.perform_later(@result, name: "A") + job = SolidQueue::Job.find_by!(active_job_id: NonOverlappingUpdateResultJob.perform_later(@result, name: "A").job_id) wait_for_jobs_to_finish_for(5.seconds) assert_no_unfinished_jobs diff --git a/test/models/solid_queue/claimed_execution_test.rb b/test/models/solid_queue/claimed_execution_test.rb index dd29b0ce..95b636f0 100644 --- a/test/models/solid_queue/claimed_execution_test.rb +++ b/test/models/solid_queue/claimed_execution_test.rb @@ -170,7 +170,7 @@ class SolidQueue::ClaimedExecutionTest < ActiveSupport::TestCase test "dispatch job with missing class and concurrency key skips concurrency controls" do job = create_job_with_missing_class(concurrency_key: "test_key") - assert_not job.concurrency_limited? + assert_nil job.job_class job.prepare_for_execution diff --git a/test/models/solid_queue/concurrency_lock_test.rb b/test/models/solid_queue/concurrency_lock_test.rb new file mode 100644 index 00000000..36487b6b --- /dev/null +++ b/test/models/solid_queue/concurrency_lock_test.rb @@ -0,0 +1,202 @@ +# frozen_string_literal: true + +require "test_helper" + +class SolidQueue::ConcurrencyLockTest < ActiveSupport::TestCase + class NonOverlappingJob < ActiveJob::Base + limits_concurrency key: "concurrency-lock-test", duration: 1.minute + + def perform(name) + end + end + + setup do + @process = SolidQueue::Process.register(kind: "Worker", pid: 42, name: "worker-123") + end + + # From the reproduction in https://github.com/rails/solid_queue/issues/827 + test "a job that outlived its lock does not release the lock a later job acquired" do + a = claim(NonOverlappingJob.perform_later("A")) + + travel_to 2.minutes.from_now + SolidQueue::Semaphore.expired.delete_all + + b = claim(NonOverlappingJob.perform_later("B")) + c = job_for(NonOverlappingJob.perform_later("C")) + assert c.blocked? + + a.perform + + assert b.job.reload.claimed? + assert c.reload.blocked? + end + + test "a job that outlived its lock does not leave the key over its limit" do + a = claim(NonOverlappingJob.perform_later("A")) + + travel_to 2.minutes.from_now + SolidQueue::Semaphore.expired.delete_all + + b = claim(NonOverlappingJob.perform_later("B")) + c = job_for(NonOverlappingJob.perform_later("C")) + d = job_for(NonOverlappingJob.perform_later("D")) + + a.perform + b.perform + claim_job(c) + + assert d.reload.blocked? + end + + test "a job that got its slot gives it back even if its class is missing when it runs" do + a = claim(NonOverlappingJob.perform_later("A")) + b = job_for(NonOverlappingJob.perform_later("B")) + assert b.blocked? + + # A worker on a release without the class claims and runs A + a.job.update_column(:class_name, "RemovedJobClass") + a = SolidQueue::ClaimedExecution.find(a.id) + + assert_raises(SolidQueue::Job::ClassMissingError) { a.perform } + + assert b.reload.ready? + assert_equal 0, semaphore.value + end + + test "a job that skipped its limits because its class was missing doesn't give back a slot" do + a = claim(NonOverlappingJob.perform_later("A")) + b = job_for(NonOverlappingJob.perform_later("B")) + + # Dispatched by a process without the class, so it takes no slot + x = SolidQueue::Job.create!(queue_name: "default", class_name: "RemovedJobClass", active_job_id: SecureRandom.uuid, + arguments: { "job_class" => "RemovedJobClass", "arguments" => [ "X" ] }, concurrency_key: a.job.concurrency_key) + assert x.ready? + + # ...and run by one that has it + x.update_columns(class_name: NonOverlappingJob.name, arguments: x.arguments.merge("job_class" => NonOverlappingJob.name)) + claim_job(x).perform + + assert a.job.reload.claimed? + assert b.reload.blocked? + end + + test "a job dispatched before Solid Queue recorded its semaphore gives back its slot by key" do + a = claim(NonOverlappingJob.perform_later("A")) + b = job_for(NonOverlappingJob.perform_later("B")) + a.job.update_column(:arguments, a.job.arguments.except(SolidQueue::Job::ConcurrencyControls::CONCURRENCY_LOCK_KEY)) + a = SolidQueue::ClaimedExecution.find(a.id) + + a.perform + + assert b.reload.ready? + end + + test "a blocked job records the semaphore it gets when it's released" do + a = claim(NonOverlappingJob.perform_later("A")) + b = job_for(NonOverlappingJob.perform_later("B")) + assert a.job.acquired_semaphore_id.present? + assert_nil b.acquired_semaphore_id + + a.perform + + assert b.reload.ready? + assert_equal semaphore.id, b.acquired_semaphore_id + assert_equal 1, b.acquired_concurrency_limit + end + + test "jobs enqueued in bulk record the semaphore they get" do + ActiveJob.perform_all_later([ NonOverlappingJob.new("A"), NonOverlappingJob.new("B") ]) + a, b = SolidQueue::Job.where(class_name: NonOverlappingJob.name).order(:id).to_a + + assert a.ready? + assert_equal semaphore.id, a.acquired_semaphore_id + assert b.blocked? + assert_nil b.acquired_semaphore_id + end + + test "jobs sharing a semaphore give their slots back to it" do + a = claim(TwoAtATimeJob.perform_later("A")) + b = claim(TwoAtATimeJob.perform_later("B")) + c = job_for(TwoAtATimeJob.perform_later("C")) + + assert_equal [ semaphore.id ] * 2, [ a, b ].map { |execution| execution.job.acquired_semaphore_id } + assert_equal 2, a.job.acquired_concurrency_limit + assert c.blocked? + + a.perform + + assert c.reload.ready? + assert_equal 0, semaphore.value + end + + test "discarding a ready job gives its slot back" do + a = job_for(NonOverlappingJob.perform_later("A")) + b = job_for(NonOverlappingJob.perform_later("B")) + + a.discard + + assert b.reload.ready? + end + + test "discarding ready jobs in bulk gives their slots back" do + a = job_for(NonOverlappingJob.perform_later("A")) + b = job_for(NonOverlappingJob.perform_later("B")) + + SolidQueue::ReadyExecution.discard_all_from_jobs([ a ]) + + assert b.reload.ready? + end + + test "a retried job records the semaphore it gets on the retry" do + a = claim(NonOverlappingJob.perform_later("A")) + a.failed_with(ExpectedTestError.new) + first_semaphore_id = a.job.reload.acquired_semaphore_id + + travel_to 2.minutes.from_now + SolidQueue::Semaphore.expired.delete_all + + a.job.failed_execution.retry + + assert a.job.reload.ready? + assert_not_equal first_semaphore_id, a.job.acquired_semaphore_id + assert_equal semaphore.id, a.job.acquired_semaphore_id + end + + test "a job records the limit it got its slot with, even when its class doesn't set one" do + # A job that kept its concurrency key after its class dropped limits_concurrency + a = SolidQueue::Job.create!(queue_name: "default", class_name: AddToBufferJob.name, active_job_id: SecureRandom.uuid, + arguments: { "job_class" => AddToBufferJob.name, "arguments" => [ 1 ] }, concurrency_key: "concurrency-lock-test") + assert_equal 1, a.acquired_concurrency_limit + + # ...and then runs where the class is missing + a.update_columns(class_name: "RemovedJobClass") + assert_raises(SolidQueue::Job::ClassMissingError) { claim_job(a).perform } + + assert_equal 1, semaphore.value + end + + private + class TwoAtATimeJob < ActiveJob::Base + limits_concurrency to: 2, key: "concurrency-lock-test", duration: 1.minute + + def perform(name) + end + end + + def semaphore + SolidQueue::Semaphore.sole + end + + def job_for(active_job) + SolidQueue::Job.find_by!(active_job_id: active_job.job_id) + end + + def claim(active_job) + claim_job(job_for(active_job)) + end + + def claim_job(job) + SolidQueue::ReadyExecution.claim(job.queue_name, 1, @process.id) + SolidQueue::ClaimedExecution.find_by!(job_id: job.id) + end +end diff --git a/test/models/solid_queue/semaphore_test.rb b/test/models/solid_queue/semaphore_test.rb index 2877afd2..a26e1528 100644 --- a/test/models/solid_queue/semaphore_test.rb +++ b/test/models/solid_queue/semaphore_test.rb @@ -16,6 +16,29 @@ def perform(key) @result = JobResult.create!(queue_name: "default") end + test "a job records the semaphore it created, not one deleted after its transaction began" do + skip "Only MySQL reads a deleted row from the transaction's snapshot" unless SolidQueue::Record.connection.adapter_name.downcase.match?(/mysql|trilogy/) + + NonOverlappingUpdateResultJob.perform_later(@result) + expired = SolidQueue::Semaphore.sole + + SolidQueue::Record.transaction do + # Take the transaction's snapshot, as dispatching does when it reads its batch of jobs + SolidQueue::Job.count + + # Maintenance deletes the expired semaphore from another connection + Thread.new do + SolidQueue::Record.connection_pool.with_connection { SolidQueue::Semaphore.where(id: expired.id).delete_all } + end.join + + job = SolidQueue::Job.enqueue(NonOverlappingUpdateResultJob.new(@result)) + + assert job.ready? + assert_not_equal expired.id, job.acquired_semaphore_id + assert_equal SolidQueue::Semaphore.lock.find_by!(key: job.concurrency_key).id, job.acquired_semaphore_id + end + end + test "wait acquires a row lock that blocks concurrent signal" do skip_on_sqlite @@ -79,7 +102,7 @@ def perform(key) class_name: "NonOverlappingUpdateResultJob", concurrency_key: concurrency_key, active_job_id: uuid, - arguments: "{}", + arguments: {}, scheduled_at: Time.current }) job_b_id = SolidQueue::Job.where(active_job_id: uuid).pick(:id)