Use FOR UPDATE SKIP LOCKED alongside advisory lock for more reliable locking

lib/good_job/job.rb

@@ -158,10 +158,7 @@ def self.queue_parser(string)
     def self.perform_with_advisory_lock
       unfinished.priority_ordered.only_scheduled.limit(1).with_advisory_lock do |good_jobs|
         good_job = good_jobs.first
-        # TODO: Determine why some records are fetched without an advisory lock at all
-        break unless good_job&.executable?
-
-        good_job.perform
+        good_job&.perform
       end
     end
 
@@ -241,12 +238,6 @@ def perform
       result
     end
 
-    # Tests whether this job is safe to be executed by this thread.
-    # @return [Boolean]
-    def executable?
-      self.class.unscoped.unfinished.owns_advisory_locked.exists?(id: id)
-    end
-
     private
 
     # @return [ExecutionResult]

lib/good_job/lockable.rb

@@ -41,9 +41,14 @@ module Lockable
 
         composed_cte = Arel::Nodes::As.new(cte_table, Arel::Nodes::SqlLiteral.new([cte_type, "(", cte_query.to_sql, ")"].join(' ')))
 
+        # In addition to an advisory lock, there is also a FOR UPDATE SKIP LOCKED
+        # because this causes the query to skip jobs that were completed (and deleted)
+        # by another session in the time since the table snapshot was taken.
+        # In rare cases under high concurrency levels, leaving this out can result in double executions.
         query = cte_table.project(cte_table[:id])
                          .with(composed_cte)
                          .where(Arel.sql(sanitize_sql_for_conditions(["pg_try_advisory_lock(('x' || substr(md5(:table_name || #{connection.quote_table_name(cte_table.name)}.#{quoted_primary_key}::text), 1, 16))::bit(64)::bigint)", { table_name: table_name }])))
+                         .lock(Arel.sql("FOR UPDATE SKIP LOCKED"))
 
         limit = original_query.arel.ast.limit
         query.limit = limit.value if limit.present?

spec/integration/scheduler_spec.rb

@@ -38,19 +38,22 @@ def perform(*args, **kwargs)
     end), transfer_nested_constants: true
   end
 
-  let(:adapter) { GoodJob::Adapter.new }
+  let(:adapter) { GoodJob::Adapter.new(execution_mode: :external) }
 
   context 'when there are a large number of jobs' do
-    let(:number_of_jobs) { 500 }
+    let(:number_of_jobs) { 1_000 }
     let(:max_threads) { 5 }
 
     let!(:good_jobs) do
-      number_of_jobs.times do |i|
-        ExampleJob.perform_later(i)
+      GoodJob::Job.transaction do
+        number_of_jobs.times do |i|
+          ExampleJob.perform_later(i)
+        end
       end
     end
 
     it 'pops items off of the queue and runs them' do
+      puts "TEST STARTS"
       performer = GoodJob::JobPerformer.new('*')
       scheduler = GoodJob::Scheduler.new(performer, max_threads: max_threads)
       max_threads.times { scheduler.create_thread }

spec/lib/good_job/job_spec.rb

@@ -177,30 +177,6 @@ def perform(result_value = nil, raise_error: false)
     end
   end
 
-  describe '#executable?' do
-    let(:good_job) { described_class.create! }
-
-    it 'is true when locked' do
-      good_job.with_advisory_lock do
-        expect(good_job.executable?).to eq true
-      end
-    end
-
-    it 'is false when job no longer exists' do
-      good_job.with_advisory_lock do
-        good_job.destroy!
-        expect(good_job.executable?).to eq false
-      end
-    end
-
-    it 'is false when the job has finished' do
-      good_job.with_advisory_lock do
-        good_job.update! finished_at: Time.current
-        expect(good_job.executable?).to eq false
-      end
-    end
-  end
-
   describe '#perform' do
     let(:active_job) { ExampleJob.new("a string") }
     let!(:good_job) { described_class.enqueue(active_job) }

spec/lib/good_job/lockable_spec.rb

@@ -35,6 +35,7 @@
           FROM "rows"
           WHERE pg_try_advisory_lock(('x' || substr(md5('good_jobs' || "rows"."id"::text), 1, 16))::bit(64)::bigint)
           LIMIT 2
+          FOR UPDATE SKIP LOCKED
         )
         ORDER BY "good_jobs"."priority" DESC
       SQL