Rename Semaphore.concurrency_key to Semaphore.key and simplify method sigantures

We can simply pass a job, since both Active Job and Solid Queue's Job respond to the
same methods, required for the semaphore to be waited on, or signaled. The "concurrency"
part of the "concurrency_key" attribute is redundant.

Author: rosa

app/models/solid_queue/blocked_execution.rb

@@ -2,7 +2,7 @@ module SolidQueue
   class BlockedExecution < SolidQueue::Execution
     assume_attributes_from_job :concurrency_key
 
-    has_one :semaphore, foreign_key: :concurrency_key, primary_key: :concurrency_key
+    has_one :semaphore, foreign_key: :key, primary_key: :concurrency_key
 
     scope :releasable, -> { left_outer_joins(:semaphore).merge(Semaphore.available.or(Semaphore.where(id: nil))) }
     scope :ordered, -> { order(priority: :asc) }
@@ -36,7 +36,7 @@ def release
 
     private
       def acquire_concurrency_lock
-        Semaphore.wait_for(concurrency_key, job.concurrency_limit, job.concurrency_limit_duration)
+        Semaphore.wait(job)
       end
 
       def promote_to_ready

app/models/solid_queue/job/concurrency_controls.rb

@@ -19,13 +19,13 @@ def unblock_blocked_jobs
         def acquire_concurrency_lock
           return true unless concurrency_limited?
 
-          Semaphore.wait_for(concurrency_key, concurrency_limit, concurrency_limit_duration)
+          Semaphore.wait(self)
         end
 
         def release_concurrency_lock
           return false unless concurrency_limited?
 
-          Semaphore.release(concurrency_key, concurrency_limit, concurrency_limit_duration)
+          Semaphore.signal(self)
         end
 
         def block

app/models/solid_queue/semaphore.rb

@@ -4,32 +4,32 @@ class SolidQueue::Semaphore < SolidQueue::Record
   scope :expired, -> { where(expires_at: ...Time.current)}
 
   class << self
-    def wait_for(concurrency_key, limit, duration)
-      if semaphore = find_by(concurrency_key: concurrency_key)
-        semaphore.value > 0 && attempt_decrement(concurrency_key, duration)
+    def wait(job)
+      if semaphore = find_by(key: job.concurrency_key)
+        semaphore.value > 0 && attempt_decrement(job.concurrency_key, job.concurrency_limit_duration)
       else
-        attempt_creation(concurrency_key, limit, duration)
+        attempt_creation(job.concurrency_key, job.concurrency_limit, job.concurrency_limit_duration)
       end
     end
 
-    def release(concurrency_key, limit, duration)
-      attempt_increment(concurrency_key, limit, duration)
+    def signal(job)
+      attempt_increment(job.concurrency_key, job.concurrency_limit, job.concurrency_limit_duration)
     end
 
     private
-      def attempt_creation(concurrency_key, limit, duration)
-        create!(concurrency_key: concurrency_key, value: limit - 1, expires_at: duration.from_now)
+      def attempt_creation(key, limit, duration)
+        create!(key: key, value: limit - 1, expires_at: duration.from_now)
         true
       rescue ActiveRecord::RecordNotUnique
-        attempt_decrement(concurrency_key, duration)
+        attempt_decrement(key, duration)
       end
 
-      def attempt_decrement(concurrency_key, duration)
-        available.where(concurrency_key: concurrency_key).update_all([ "value = value - 1, expires_at = ?", duration.from_now ]) > 0
+      def attempt_decrement(key, duration)
+        available.where(key: key).update_all([ "value = value - 1, expires_at = ?", duration.from_now ]) > 0
       end
 
-      def attempt_increment(concurrency_key, limit, duration)
-        where("value < ?", limit).where(concurrency_key: concurrency_key).update_all([ "value = value + 1, expires_at = ?", duration.from_now ]) > 0
+      def attempt_increment(key, limit, duration)
+        where("value < ?", limit).where(key: key).update_all([ "value = value + 1, expires_at = ?", duration.from_now ]) > 0
       end
   end
 end

db/migrate/20231103204612_create_solid_queue_concurrency_controls.rb

@@ -17,7 +17,7 @@ def change
     end
 
     create_table :solid_queue_semaphores do |t|
-      t.string :concurrency_key, null: false, index: { unique: true }
+      t.string :key, null: false, index: { unique: true }
       t.integer :value, null: false, default: 1
       t.datetime :expires_at, null: false, index: true
 

test/dummy/db/schema.rb

@@ -96,13 +96,13 @@
   end
 
   create_table "solid_queue_semaphores", charset: "utf8mb4", collation: "utf8mb4_0900_ai_ci", force: :cascade do |t|
-    t.string "concurrency_key", null: false
+    t.string "key", null: false
     t.integer "value", default: 1, null: false
     t.datetime "expires_at", null: false
     t.datetime "created_at", null: false
     t.datetime "updated_at", null: false
-    t.index ["concurrency_key"], name: "index_solid_queue_semaphores_on_concurrency_key", unique: true
     t.index ["expires_at"], name: "index_solid_queue_semaphores_on_expires_at"
+    t.index ["key"], name: "index_solid_queue_semaphores_on_key", unique: true
   end
 
 end

test/integration/concurrency_controls_test.rb

@@ -91,7 +91,7 @@ class ConcurrencyControlsTest < ActiveSupport::TestCase
 
     # Lock the semaphore so we can enqueue jobs and leave them blocked
     skip_active_record_query_cache do
-      assert SolidQueue::Semaphore.wait_for(job.concurrency_key, job.concurrency_limit, job.concurrency_limit_duration)
+      assert SolidQueue::Semaphore.wait(job)
     end
 
     # Now enqueue more jobs under that same key. They'll be all locked. Use priorities
@@ -104,7 +104,7 @@ class ConcurrencyControlsTest < ActiveSupport::TestCase
 
     # Then unlock the semaphore: this would be as if the first job had released
     # the semaphore but hadn't unblocked any jobs
-    assert SolidQueue::Semaphore.release(job.concurrency_key, job.concurrency_limit, job.concurrency_limit_duration)
+    assert SolidQueue::Semaphore.signal(job)
 
     # And wait for workers to release the jobs
     wait_for_jobs_to_finish_for(2.seconds)
@@ -123,7 +123,7 @@ class ConcurrencyControlsTest < ActiveSupport::TestCase
 
     # Lock the semaphore so we can enqueue jobs and leave them blocked
     skip_active_record_query_cache do
-      assert SolidQueue::Semaphore.wait_for(job.concurrency_key, job.concurrency_limit, job.concurrency_limit_duration)
+      assert SolidQueue::Semaphore.wait(job)
     end
 
     # Now enqueue more jobs under that same key. They'll be all locked
@@ -134,7 +134,7 @@ class ConcurrencyControlsTest < ActiveSupport::TestCase
     end
 
     # Simulate semaphore expiration
-    SolidQueue::Semaphore.find_by(concurrency_key: job.concurrency_key).update(expires_at: 1.hour.ago)
+    SolidQueue::Semaphore.find_by(key: job.concurrency_key).update(expires_at: 1.hour.ago)
 
     # And wait for scheduler to release the jobs
     wait_for_jobs_to_finish_for(2.seconds)