From 7ad4dfa13aa3413fcdf1d049d9abe60aee9d6337 Mon Sep 17 00:00:00 2001
From: Rosa Gutierrez <rosa@37signals.com>
Date: Tue, 3 Nov 2026 12:47:47 +0100
Subject: [PATCH 1/5] Give concurrency slots back only to the semaphore they
 came from

A semaphore counts free slots but doesn't know which jobs hold them, so
any job that gave a slot back by key counted as a holder:

- A job that ran past its duration gave its slot back to the semaphore
  created after its own expired, letting another job start while the
  new holder was still running (#827).
- A job dispatched while its class was missing skipped its limits, but
  gave a slot back if it later ran where its class loads.
- A job whose class was missing when it ran couldn't give back the slot
  it got when it was dispatched (#825).

Now a job records the id of the semaphore it got a slot on, and its
limit, in its serialized Active Job data, and gives the slot back only
to that semaphore. A job that skips its limits records that it holds
no slot. Jobs that got their slot before this change have nothing
recorded and keep giving it back by key, until 2.0.

The first two tests come from the reproduction in #827.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 app/models/solid_queue/blocked_execution.rb   |   9 +-
 .../solid_queue/job/concurrency_controls.rb   |  49 +++++--
 app/models/solid_queue/job/executable.rb      |   3 +-
 app/models/solid_queue/semaphore.rb           |  29 +++-
 test/integration/concurrency_controls_test.rb |   4 +-
 .../solid_queue/concurrency_lock_test.rb      | 134 ++++++++++++++++++
 test/models/solid_queue/semaphore_test.rb     |   2 +-
 7 files changed, 206 insertions(+), 24 deletions(-)
 create mode 100644 test/models/solid_queue/concurrency_lock_test.rb

diff --git a/app/models/solid_queue/blocked_execution.rb b/app/models/solid_queue/blocked_execution.rb
index a84660a46..fd0853498 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 4acec12a5..01e3d72d6 100644
--- a/app/models/solid_queue/job/concurrency_controls.rb
+++ b/app/models/solid_queue/job/concurrency_controls.rb
@@ -5,12 +5,15 @@ 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
+        delegate :concurrency_limit, to: :job_class
 
-        before_destroy :unblock_next_blocked_job, if: -> { concurrency_limited? && ready? }
+        before_destroy :unblock_next_blocked_job, if: -> { concurrency_key.present? && ready? }
       end
 
       class_methods do
@@ -29,6 +32,31 @@ def concurrency_limited?
         concurrency_key.present? && job_class.present?
       end
 
+      def concurrency_duration
+        job_class ? job_class.concurrency_duration : SolidQueue.default_concurrency_control_period
+      end
+
+      def acquire_concurrency_lock
+        if concurrency_limited?
+          if semaphore_id = Semaphore.wait(self)
+            record_concurrency_lock(semaphore_id:, limit: concurrency_limit)
+          end
+        else
+          # 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) if concurrency_key.present?
+          true
+        end
+      end
+
+      def concurrency_lock
+        arguments[CONCURRENCY_LOCK_KEY]
+      end
+
+      def concurrency_lock_recorded?
+        arguments.key?(CONCURRENCY_LOCK_KEY)
+      end
+
       def blocked?
         blocked_execution.present?
       end
@@ -42,16 +70,19 @@ def concurrency_on_conflict
           job_class.concurrency_on_conflict.to_s.inquiry
         end
 
-        def acquire_concurrency_lock
-          return true unless concurrency_limited?
-
-          Semaphore.wait(self)
+        def record_concurrency_lock(semaphore_id:, limit: nil)
+          arguments[CONCURRENCY_LOCK_KEY] = { "semaphore_id" => semaphore_id, "limit" => limit }
+          update_column :arguments, arguments
+          true
         end
 
         def release_concurrency_lock
-          return false unless concurrency_limited?
-
-          Semaphore.signal(self)
+          if concurrency_lock_recorded?
+            concurrency_lock["semaphore_id"].present? && Semaphore.signal(self)
+          else
+            # Jobs dispatched before Solid Queue recorded their semaphore
+            concurrency_limited? && Semaphore.signal(self)
+          end
         end
 
         def handle_concurrency_conflict
diff --git a/app/models/solid_queue/job/executable.rb b/app/models/solid_queue/job/executable.rb
index a3860e1fe..e3c70902d 100644
--- a/app/models/solid_queue/job/executable.rb
+++ b/app/models/solid_queue/job/executable.rb
@@ -26,7 +26,8 @@ def prepare_all_for_execution(jobs)
         end
 
         def dispatch_all(jobs)
-          with_concurrency_limits, without_concurrency_limits = jobs.partition(&:concurrency_limited?)
+          # Jobs with a concurrency key go one by one, so they record the slot they get, or that they skipped it
+          with_concurrency_limits, without_concurrency_limits = jobs.partition { |job| job.concurrency_key.present? }
 
           dispatch_all_at_once(without_concurrency_limits)
           dispatch_all_one_by_one(with_concurrency_limits)
diff --git a/app/models/solid_queue/semaphore.rb b/app/models/solid_queue/semaphore.rb
index da2ea4c3b..a4e859654 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
@@ -48,14 +49,18 @@ def signal
 
         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
+          acquired_semaphore_id if limit > 1 && attempt_decrement
+        end
+
+        def acquired_semaphore_id
+          Semaphore.where(key: key).pick(:id)
         end
 
         def attempt_decrement
@@ -63,7 +68,18 @@ def 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_for_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.concurrency_lock_recorded?
+            Semaphore.where(id: job.concurrency_lock["semaphore_id"])
+          else
+            Semaphore.where(key: key)
+          end
         end
 
         def key
@@ -77,6 +93,11 @@ def expires_at
         def limit
           job.concurrency_limit || 1
         end
+
+        # Recorded when the job got its slot, so it's known even if the job's class no longer loads
+        def limit_for_release
+          job.concurrency_lock&.dig("limit") || limit
+        end
     end
   end
 end
diff --git a/test/integration/concurrency_controls_test.rb b/test/integration/concurrency_controls_test.rb
index 147100b3b..b6383c2e9 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/concurrency_lock_test.rb b/test/models/solid_queue/concurrency_lock_test.rb
new file mode 100644
index 000000000..535fbe9b6
--- /dev/null
+++ b/test/models/solid_queue/concurrency_lock_test.rb
@@ -0,0 +1,134 @@
+# 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.concurrency_lock["semaphore_id"].present?
+    assert_not b.concurrency_lock_recorded?
+
+    a.perform
+
+    assert b.reload.ready?
+    assert_equal semaphore.id, b.concurrency_lock["semaphore_id"]
+    assert_equal 1, b.concurrency_lock["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.concurrency_lock["semaphore_id"]
+    assert b.blocked?
+    assert_not b.concurrency_lock_recorded?
+  end
+
+  private
+    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 2877afd27..e201ccaaf 100644
--- a/test/models/solid_queue/semaphore_test.rb
+++ b/test/models/solid_queue/semaphore_test.rb
@@ -79,7 +79,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)

From dd5437811624bdeaa9fa86c36e274749fd924bf4 Mon Sep 17 00:00:00 2001
From: Rosa Gutierrez <rosa@37signals.com>
Date: Tue, 3 Nov 2026 16:09:37 +0100
Subject: [PATCH 2/5] Read the acquired semaphore with a lock, and record the
 limit used

On MySQL with REPEATABLE READ, a plain read after taking a slot uses the
transaction's snapshot, which can still include a semaphore that expired
and was deleted after the snapshot was taken, so the job could record
that one instead of the semaphore it just created. A locking read sees
the latest committed rows.

A job that keeps its concurrency key after its class drops
limits_concurrency takes its slot with a limit of 1, so record that
rather than nil, which failed to give the slot back if the class was
also missing when the job ran.

Also cover jobs sharing a semaphore, discarding ready jobs one by one
and in bulk, and retried jobs, and explain the change in UPGRADING.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 UPGRADING.md                                  |  7 ++
 .../solid_queue/job/concurrency_controls.rb   |  2 +-
 app/models/solid_queue/semaphore.rb           |  4 +-
 .../solid_queue/concurrency_lock_test.rb      | 68 +++++++++++++++++++
 4 files changed, 79 insertions(+), 2 deletions(-)

diff --git a/UPGRADING.md b/UPGRADING.md
index 2eecfd0c4..766c56230 100644
--- a/UPGRADING.md
+++ b/UPGRADING.md
@@ -1,3 +1,10 @@
+# Upgrading to version 1.8.x
+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, let the concurrency-limited jobs that were ready or running when you upgraded to 1.8 finish. If any haven't, their slots stay taken until their semaphores expire, after `duration`, and then the jobs waiting for them are released as usual.
+
 # 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/job/concurrency_controls.rb b/app/models/solid_queue/job/concurrency_controls.rb
index 01e3d72d6..4f5264a91 100644
--- a/app/models/solid_queue/job/concurrency_controls.rb
+++ b/app/models/solid_queue/job/concurrency_controls.rb
@@ -39,7 +39,7 @@ def concurrency_duration
       def acquire_concurrency_lock
         if concurrency_limited?
           if semaphore_id = Semaphore.wait(self)
-            record_concurrency_lock(semaphore_id:, limit: concurrency_limit)
+            record_concurrency_lock(semaphore_id:, limit: concurrency_limit || 1)
           end
         else
           # A job whose class doesn't load skips its limits. Record that it holds no slot,
diff --git a/app/models/solid_queue/semaphore.rb b/app/models/solid_queue/semaphore.rb
index a4e859654..c7805dba3 100644
--- a/app/models/solid_queue/semaphore.rb
+++ b/app/models/solid_queue/semaphore.rb
@@ -59,8 +59,10 @@ def check_limit_or_decrement
           acquired_semaphore_id if limit > 1 && attempt_decrement
         end
 
+        # 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.where(key: key).pick(:id)
+          Semaphore.lock.where(key: key).pick(:id)
         end
 
         def attempt_decrement
diff --git a/test/models/solid_queue/concurrency_lock_test.rb b/test/models/solid_queue/concurrency_lock_test.rb
index 535fbe9b6..66095e438 100644
--- a/test/models/solid_queue/concurrency_lock_test.rb
+++ b/test/models/solid_queue/concurrency_lock_test.rb
@@ -114,7 +114,75 @@ def perform(name)
     assert_not b.concurrency_lock_recorded?
   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.concurrency_lock["semaphore_id"] }
+    assert_equal 2, a.job.concurrency_lock["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.concurrency_lock["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.concurrency_lock["semaphore_id"]
+    assert_equal semaphore.id, a.job.concurrency_lock["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.concurrency_lock["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

From d72e15d9fe827b797773ae9d9c02b012f09a4502 Mon Sep 17 00:00:00 2001
From: Rosa Gutierrez <rosa@37signals.com>
Date: Tue, 3 Nov 2026 16:09:47 +0100
Subject: [PATCH 3/5] Place the semaphore note under 1.7.2 in UPGRADING

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 UPGRADING.md | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/UPGRADING.md b/UPGRADING.md
index 766c56230..3830972e3 100644
--- a/UPGRADING.md
+++ b/UPGRADING.md
@@ -1,9 +1,9 @@
-# Upgrading to version 1.8.x
+# 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, let the concurrency-limited jobs that were ready or running when you upgraded to 1.8 finish. If any haven't, their slots stay taken until their semaphores expire, after `duration`, and then the jobs waiting for them are released as usual.
+Solid Queue 2.0 will stop giving back slots that weren't recorded. Before upgrading to 2.0, let the concurrency-limited jobs that were ready or running when you upgraded to 1.7.2 finish. If any haven't, their slots stay taken until their semaphores expire, after `duration`, and then the jobs waiting for them are released as usual.
 
 # 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:

From 126da2f2b3ca7d55c391ed50e3f90762e53d9a5b Mon Sep 17 00:00:00 2001
From: Rosa Gutierrez <rosa@37signals.com>
Date: Tue, 3 Nov 2026 16:21:18 +0100
Subject: [PATCH 4/5] Test that a job records the semaphore it created on
 MySQL, and correct the upgrade note

A plain read inside a REPEATABLE READ transaction can still see a
semaphore deleted after the transaction's snapshot, so without the
locking read the job would record that one. The test fails without it.

A slot that isn't given back comes back only when its semaphore expires,
and every job taking or giving back a slot for the same key pushes that
out, so on a busy key with a limit above 1 it may not come back for a
long time. Say so in UPGRADING instead of promising it comes back after
duration.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 UPGRADING.md                              |  2 +-
 test/models/solid_queue/semaphore_test.rb | 23 +++++++++++++++++++++++
 2 files changed, 24 insertions(+), 1 deletion(-)

diff --git a/UPGRADING.md b/UPGRADING.md
index 3830972e3..746721827 100644
--- a/UPGRADING.md
+++ b/UPGRADING.md
@@ -3,7 +3,7 @@ Jobs with [concurrency controls](README.md#concurrency-controls) now remember wh
 
 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, let the concurrency-limited jobs that were ready or running when you upgraded to 1.7.2 finish. If any haven't, their slots stay taken until their semaphores expire, after `duration`, and then the jobs waiting for them are released as usual.
+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/test/models/solid_queue/semaphore_test.rb b/test/models/solid_queue/semaphore_test.rb
index e201ccaaf..2a7ab4380 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.concurrency_lock["semaphore_id"]
+      assert_equal SolidQueue::Semaphore.lock.find_by!(key: job.concurrency_key).id, job.concurrency_lock["semaphore_id"]
+    end
+  end
+
   test "wait acquires a row lock that blocks concurrent signal" do
     skip_on_sqlite
 

From 0c060818727f5fb90efb5f52fca82538e86a1a00 Mon Sep 17 00:00:00 2001
From: Rosa Gutierrez <rosa@37signals.com>
Date: Tue, 3 Nov 2026 19:53:36 +0100
Subject: [PATCH 5/5] Let the job answer for the concurrency slot it holds

concurrency_limited? now just means the job has a concurrency key, as on
the Active Job side, and the code checks for a missing class where that
matters. The job exposes the semaphore it got its slot from and the
limit it got it with, so the semaphore asks for those instead of
reading the recorded data itself.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 .../solid_queue/job/concurrency_controls.rb   | 84 +++++++++++--------
 app/models/solid_queue/job/executable.rb      |  3 +-
 app/models/solid_queue/semaphore.rb           | 30 +++----
 .../solid_queue/claimed_execution_test.rb     |  2 +-
 .../solid_queue/concurrency_lock_test.rb      | 24 +++---
 test/models/solid_queue/semaphore_test.rb     |  4 +-
 6 files changed, 81 insertions(+), 66 deletions(-)

diff --git a/app/models/solid_queue/job/concurrency_controls.rb b/app/models/solid_queue/job/concurrency_controls.rb
index 4f5264a91..ccf985c05 100644
--- a/app/models/solid_queue/job/concurrency_controls.rb
+++ b/app/models/solid_queue/job/concurrency_controls.rb
@@ -11,9 +11,7 @@ module ConcurrencyControls
       included do
         has_one :blocked_execution
 
-        delegate :concurrency_limit, to: :job_class
-
-        before_destroy :unblock_next_blocked_job, if: -> { concurrency_key.present? && ready? }
+        before_destroy :unblock_next_blocked_job, if: -> { concurrency_limited? && ready? }
       end
 
       class_methods do
@@ -28,33 +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
 
-      def concurrency_duration
-        job_class ? job_class.concurrency_duration : SolidQueue.default_concurrency_control_period
+      # 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 acquire_concurrency_lock
-        if concurrency_limited?
-          if semaphore_id = Semaphore.wait(self)
-            record_concurrency_lock(semaphore_id:, limit: concurrency_limit || 1)
-          end
-        else
-          # 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) if concurrency_key.present?
-          true
-        end
+      def concurrency_duration
+        job_class ? job_class.concurrency_duration : SolidQueue.default_concurrency_control_period
       end
 
-      def concurrency_lock
-        arguments[CONCURRENCY_LOCK_KEY]
+      # The semaphore the job got its concurrency slot from
+      def acquired_semaphore_id
+        concurrency_lock&.dig("semaphore_id")
       end
 
-      def concurrency_lock_recorded?
-        arguments.key?(CONCURRENCY_LOCK_KEY)
+      # The concurrency limit the job had when it got its slot
+      def acquired_concurrency_limit
+        concurrency_lock&.dig("limit")
       end
 
       def blocked?
@@ -66,8 +70,25 @@ 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 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
+
+        def concurrency_lock_recorded?
+          arguments.key?(CONCURRENCY_LOCK_KEY)
+        end
+
+        def release_next_blocked_job
+          BlockedExecution.release_one(concurrency_key)
         end
 
         def record_concurrency_lock(semaphore_id:, limit: nil)
@@ -76,13 +97,8 @@ def record_concurrency_lock(semaphore_id:, limit: nil)
           true
         end
 
-        def release_concurrency_lock
-          if concurrency_lock_recorded?
-            concurrency_lock["semaphore_id"].present? && Semaphore.signal(self)
-          else
-            # Jobs dispatched before Solid Queue recorded their semaphore
-            concurrency_limited? && Semaphore.signal(self)
-          end
+        def concurrency_lock
+          arguments[CONCURRENCY_LOCK_KEY]
         end
 
         def handle_concurrency_conflict
@@ -93,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/job/executable.rb b/app/models/solid_queue/job/executable.rb
index e3c70902d..a3860e1fe 100644
--- a/app/models/solid_queue/job/executable.rb
+++ b/app/models/solid_queue/job/executable.rb
@@ -26,8 +26,7 @@ def prepare_all_for_execution(jobs)
         end
 
         def dispatch_all(jobs)
-          # Jobs with a concurrency key go one by one, so they record the slot they get, or that they skipped it
-          with_concurrency_limits, without_concurrency_limits = jobs.partition { |job| job.concurrency_key.present? }
+          with_concurrency_limits, without_concurrency_limits = jobs.partition(&:concurrency_limited?)
 
           dispatch_all_at_once(without_concurrency_limits)
           dispatch_all_one_by_one(with_concurrency_limits)
diff --git a/app/models/solid_queue/semaphore.rb b/app/models/solid_queue/semaphore.rb
index c7805dba3..4fa6768d7 100644
--- a/app/models/solid_queue/semaphore.rb
+++ b/app/models/solid_queue/semaphore.rb
@@ -47,6 +47,10 @@ 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)
             acquired_semaphore_id
@@ -55,35 +59,36 @@ def attempt_creation
           end
         end
 
-        def check_limit_or_decrement
-          acquired_semaphore_id if limit > 1 && attempt_decrement
-        end
-
         # 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_to_release.where(value: ...limit_for_release).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.concurrency_lock_recorded?
-            Semaphore.where(id: job.concurrency_lock["semaphore_id"])
+          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
           job.concurrency_key
         end
@@ -93,12 +98,7 @@ def expires_at
         end
 
         def limit
-          job.concurrency_limit || 1
-        end
-
-        # Recorded when the job got its slot, so it's known even if the job's class no longer loads
-        def limit_for_release
-          job.concurrency_lock&.dig("limit") || limit
+          job.concurrency_limit
         end
     end
   end
diff --git a/test/models/solid_queue/claimed_execution_test.rb b/test/models/solid_queue/claimed_execution_test.rb
index dd29b0ceb..95b636f01 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
index 66095e438..36487b6bb 100644
--- a/test/models/solid_queue/concurrency_lock_test.rb
+++ b/test/models/solid_queue/concurrency_lock_test.rb
@@ -94,14 +94,14 @@ def perform(name)
   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.concurrency_lock["semaphore_id"].present?
-    assert_not b.concurrency_lock_recorded?
+    assert a.job.acquired_semaphore_id.present?
+    assert_nil b.acquired_semaphore_id
 
     a.perform
 
     assert b.reload.ready?
-    assert_equal semaphore.id, b.concurrency_lock["semaphore_id"]
-    assert_equal 1, b.concurrency_lock["limit"]
+    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
@@ -109,9 +109,9 @@ def perform(name)
     a, b = SolidQueue::Job.where(class_name: NonOverlappingJob.name).order(:id).to_a
 
     assert a.ready?
-    assert_equal semaphore.id, a.concurrency_lock["semaphore_id"]
+    assert_equal semaphore.id, a.acquired_semaphore_id
     assert b.blocked?
-    assert_not b.concurrency_lock_recorded?
+    assert_nil b.acquired_semaphore_id
   end
 
   test "jobs sharing a semaphore give their slots back to it" do
@@ -119,8 +119,8 @@ def perform(name)
     b = claim(TwoAtATimeJob.perform_later("B"))
     c = job_for(TwoAtATimeJob.perform_later("C"))
 
-    assert_equal [ semaphore.id ] * 2, [ a, b ].map { |execution| execution.job.concurrency_lock["semaphore_id"] }
-    assert_equal 2, a.job.concurrency_lock["limit"]
+    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
@@ -150,7 +150,7 @@ def perform(name)
   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.concurrency_lock["semaphore_id"]
+    first_semaphore_id = a.job.reload.acquired_semaphore_id
 
     travel_to 2.minutes.from_now
     SolidQueue::Semaphore.expired.delete_all
@@ -158,15 +158,15 @@ def perform(name)
     a.job.failed_execution.retry
 
     assert a.job.reload.ready?
-    assert_not_equal first_semaphore_id, a.job.concurrency_lock["semaphore_id"]
-    assert_equal semaphore.id, a.job.concurrency_lock["semaphore_id"]
+    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.concurrency_lock["limit"]
+    assert_equal 1, a.acquired_concurrency_limit
 
     # ...and then runs where the class is missing
     a.update_columns(class_name: "RemovedJobClass")
diff --git a/test/models/solid_queue/semaphore_test.rb b/test/models/solid_queue/semaphore_test.rb
index 2a7ab4380..a26e15289 100644
--- a/test/models/solid_queue/semaphore_test.rb
+++ b/test/models/solid_queue/semaphore_test.rb
@@ -34,8 +34,8 @@ def perform(key)
       job = SolidQueue::Job.enqueue(NonOverlappingUpdateResultJob.new(@result))
 
       assert job.ready?
-      assert_not_equal expired.id, job.concurrency_lock["semaphore_id"]
-      assert_equal SolidQueue::Semaphore.lock.find_by!(key: job.concurrency_key).id, job.concurrency_lock["semaphore_id"]
+      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