Lock the rows when taking the advisory lock

lib/que/adapters/active_record_with_lock.rb

@@ -35,13 +35,24 @@ def execute(command, params = [])
       def lock_job_with_lock_database(queue, cursor)
         result = []
         loop do
-          result = Que.execute(:find_job_to_lock, [queue, cursor])
+          break_loop = false
+          Que.transaction do
+            result = Que.execute(:find_job_to_lock, [queue, cursor])
 
-          break if result.empty?
+            if result.empty?
+              break_loop = true
+              break
+            end
 
-          cursor = result.first["job_id"]
-          break if pg_try_advisory_lock?(cursor)
+            cursor = result.first["job_id"]
+            if pg_try_advisory_lock?(cursor)
+              break_loop = true
+              break
+            end
+          end
+          break if break_loop
         end
+
         result
       end
 

lib/que/sql.rb

@@ -184,7 +184,7 @@ module Que
             AND retryable = true
             AND job_id >= $2
             ORDER BY priority, run_at, job_id
-            LIMIT 1
+            for update skip locked LIMIT 1
     },
   }
   # rubocop:enable Style/MutableConstant

spec/active_record_with_lock_spec_helper.rb

@@ -4,7 +4,7 @@ class LockDatabaseRecord < ActiveRecord::Base
   establish_connection(
     adapter: "postgresql",
     host: ENV.fetch("LOCK_PGHOST", "localhost"),
-    user: ENV.fetch("LOCK_PGUSER", "postgres"),
+    user: ENV.fetch("LOCK_PGUSER", "ubuntu"),
     password: ENV.fetch("LOCK_PGPASSWORD", "password"),
     database: ENV.fetch("LOCK_PGDATABASE", "lock-test"),
     port: ENV.fetch("LOCK_PGPORT", 5434),