Make sure jobs claimed are those claimed for the given process ID

app/models/solid_queue/claimed_execution.rb

@@ -11,8 +11,8 @@ class << self
     def claiming(job_ids, process_id, &block)
       job_data = Array(job_ids).collect { |job_id| { job_id: job_id, process_id: process_id } }
 
-      insert_all(job_data)
-      where(job_id: job_ids).load.tap do |claimed|
+      insert_all!(job_data)
+      where(job_id: job_ids, process_id: process_id).load.tap do |claimed|
         block.call(claimed)
         SolidQueue.logger.info("[SolidQueue] Claimed #{claimed.size} jobs")
       end

app/models/solid_queue/ready_execution.rb

@@ -18,18 +18,19 @@ def select_and_lock(queue_relation, process_id, limit)
           return [] if limit <= 0
 
           transaction do
-            candidates = select_candidates(queue_relation, limit)
-            lock(candidates, process_id)
+            job_ids = select_candidates(queue_relation, limit)
+            lock_candidates(job_ids, process_id)
           end
         end
 
         def select_candidates(queue_relation, limit)
           queue_relation.ordered.limit(limit).lock("FOR UPDATE SKIP LOCKED").pluck(:job_id)
         end
 
-        def lock(candidates, process_id)
-          return [] if candidates.none?
-          SolidQueue::ClaimedExecution.claiming(candidates, process_id) do |claimed|
+        def lock_candidates(job_ids, process_id)
+          return [] if job_ids.none?
+
+          SolidQueue::ClaimedExecution.claiming(job_ids, process_id) do |claimed|
             where(job_id: claimed.pluck(:job_id)).delete_all
           end
         end

db/migrate/20231129215414_extend_claimed_executions_index_on_process_id.rb

@@ -0,0 +1,6 @@
+class ExtendClaimedExecutionsIndexOnProcessId < ActiveRecord::Migration[7.1]
+  def change
+    add_index :solid_queue_claimed_executions, [ :process_id, :job_id ]
+    remove_index :solid_queue_claimed_executions, :process_id
+  end
+end

test/dummy/db/schema.rb

@@ -10,7 +10,7 @@
 #
 # It's strongly recommended that you check this file into your version control system.
 
-ActiveRecord::Schema[7.1].define(version: 2023_11_15_211044) do
+ActiveRecord::Schema[7.1].define(version: 2023_11_29_215414) do
   create_table "job_results", charset: "utf8mb4", collation: "utf8mb4_0900_ai_ci", force: :cascade do |t|
     t.string "queue_name"
     t.string "status"
@@ -26,6 +26,7 @@
     t.string "concurrency_key", null: false
     t.datetime "created_at", null: false
     t.datetime "expires_at", null: false
+    t.index ["concurrency_key", "expires_at"], name: "index_solid_queue_blocked_executions_for_maintenance_2"
     t.index ["concurrency_key", "priority", "job_id"], name: "index_solid_queue_blocked_executions_for_release"
     t.index ["expires_at", "concurrency_key"], name: "index_solid_queue_blocked_executions_for_maintenance"
     t.index ["job_id"], name: "index_solid_queue_blocked_executions_on_job_id", unique: true
@@ -36,7 +37,7 @@
     t.bigint "process_id"
     t.datetime "created_at", null: false
     t.index ["job_id"], name: "index_solid_queue_claimed_executions_on_job_id", unique: true
-    t.index ["process_id"], name: "index_solid_queue_claimed_executions_on_process_id"
+    t.index ["process_id", "job_id"], name: "index_solid_queue_claimed_executions_on_process_id_and_job_id"
   end
 
   create_table "solid_queue_failed_executions", charset: "utf8mb4", collation: "utf8mb4_0900_ai_ci", force: :cascade do |t|