Prevent false worker pruning from leaking concurrency permits

README.md

@@ -373,7 +373,7 @@ The supervisor is in charge of managing these processes, and it responds to the
 
 When receiving a `QUIT` signal, if workers still have jobs in-flight, these will be returned to the queue when the processes are deregistered.
 
-If processes have no chance of cleaning up before exiting (e.g. if someone pulls a cable somewhere), in-flight jobs might remain claimed by the processes executing them. Processes send heartbeats, and the supervisor checks and prunes processes with expired heartbeats. Jobs that were claimed by processes with an expired heartbeat will be marked as failed with a `SolidQueue::Processes::ProcessPrunedError`. You can configure both the frequency of heartbeats and the threshold to consider a process dead. See the section below for this.
+If processes have no chance of cleaning up before exiting (e.g. if someone pulls a cable somewhere), in-flight jobs might remain claimed by the processes executing them. Processes send heartbeats, and the supervisor checks and prunes processes with expired heartbeats. A stale supervised process isn't pruned while its supervisor has a fresh heartbeat, because the supervisor detects actual child exits directly. It becomes eligible for heartbeat-based pruning when its supervisor is also stale. Jobs claimed by a pruned process will be marked as failed with a `SolidQueue::Processes::ProcessPrunedError`. You can configure both the frequency of heartbeats and the threshold to consider a process dead. See the section below for this.
 
 In a similar way, if a worker is terminated in any other way not initiated by the above signals (e.g. a worker is sent a `KILL` signal), jobs in progress will be marked as failed so that they can be inspected, with a `SolidQueue::Processes::ProcessExitError`. Sometimes a job in particular is responsible for this, for example, if it has a memory leak and you have a mechanism to kill processes over a certain memory threshold, so this will help identifying this kind of situation.
 

app/models/solid_queue/claimed_execution.rb

@@ -43,7 +43,6 @@ def fail_all_with(error)
         SolidQueue.instrument(:fail_many_claimed) do |payload|
           executions.each do |execution|
             execution.failed_with(error)
-            execution.unblock_next_job
           end
 
           payload[:process_ids] = executions.map(&:process_id).uniq
@@ -71,13 +70,11 @@ def perform
       failed_with(result.error)
       raise result.error
     end
-  ensure
-    unblock_next_job
   end
 
   def release
     SolidQueue.instrument(:release_claimed, job_id: job.id, process_id: process_id) do
-      transaction do
+      with_claim do
         job.dispatch_bypassing_concurrency_limits
         destroy!
       end
@@ -89,16 +86,11 @@ def discard
   end
 
   def failed_with(error)
-    transaction do
+    finalize_claim do
       job.failed_with(error)
-      destroy!
     end
   end
 
-  def unblock_next_job
-    job.unblock_next_blocked_job
-  end
-
   private
     def execute
       ActiveJob::Base.execute(job.arguments.merge("provider_job_id" => job.id))
@@ -108,9 +100,32 @@ def execute
     end
 
     def finished
-      transaction do
+      finalize_claim do
         job.finished!
+      end
+    end
+
+    def finalize_claim
+      permit_released = false
+
+      with_claim do
+        yield
         destroy!
+        # Return the permit inside the claim transaction so only the claim's
+        # owner can return it. Promoting the next blocked job is done after the
+        # transaction commits, so a failure there can't roll back the finished
+        # or failed state of a job that already ran.
+        permit_released = job.release_concurrency_permit
+      end
+
+      job.promote_next_blocked_job if permit_released
+    end
+
+    def with_claim
+      transaction do
+        return false unless self.class.unscoped.lock.find_by(id: id)
+
+        yield
       end
     end
 end

app/models/solid_queue/job/concurrency_controls.rb

@@ -20,9 +20,18 @@ def release_all_concurrency_locks(jobs)
       end
 
       def unblock_next_blocked_job
-        if release_concurrency_lock
-          release_next_blocked_job
-        end
+        promote_next_blocked_job if release_concurrency_permit
+      end
+
+      # Return this job's concurrency permit. Kept separate from promotion so the
+      # permit can be returned inside the claim transaction while promoting the
+      # next blocked job happens after that transaction commits.
+      def release_concurrency_permit
+        release_concurrency_lock
+      end
+
+      def promote_next_blocked_job
+        release_next_blocked_job
       end
 
       def concurrency_limited?

app/models/solid_queue/process/prunable.rb

@@ -6,7 +6,14 @@ module Prunable
       extend ActiveSupport::Concern
 
       included do
-        scope :prunable, -> { where(last_heartbeat_at: ..SolidQueue.process_alive_threshold.ago) }
+        scope :prunable, -> {
+          heartbeat_cutoff = SolidQueue.process_alive_threshold.ago
+          alive_processes = where(last_heartbeat_at: heartbeat_cutoff..)
+          protected_processes = where(supervisor_id: alive_processes.select(:id))
+
+          where(last_heartbeat_at: ..heartbeat_cutoff)
+            .where.not(id: protected_processes.select(:id))
+        }
       end
 
       class_methods do

lib/solid_queue/supervisor/maintenance.rb

@@ -40,6 +40,7 @@ def release_claimed_jobs_by(terminated_process, with_error:)
         wrap_in_app_executor do
           if registered_process = SolidQueue::Process.find_by(name: terminated_process.name)
             registered_process.fail_all_claimed_executions_with(with_error)
+            registered_process.deregister(pruned: true)
           end
         end
       end

test/integration/forked_processes_lifecycle_test.rb

@@ -244,8 +244,17 @@ class ForkedProcessesLifecycleTest < ActiveSupport::TestCase
     sleep(0.5.second)
     assert SolidQueue::Process.exists?(id: worker.id)
 
-    # And there's a new worker that has been registered for the background queue
-    wait_for_registered_processes(4, timeout: 5.second)
+    # The supervisor reaps the killed worker, deregisters it, and starts a
+    # replacement. The reap and the replacement are separate async events, so
+    # wait for both: first the dead row disappearing, then the fleet returning to
+    # full strength. Asserting between them catches a transient count of 2 (dead
+    # row gone, replacement not registered yet).
+    wait_while_with_timeout(5.seconds) { SolidQueue::Process.exists?(id: worker.id) }
+    wait_for_registered_processes(3, timeout: 5.seconds)
+    skip_active_record_query_cache do
+      assert_not SolidQueue::Process.exists?(id: worker.id)
+      assert_equal 3, SolidQueue::Process.count
+    end
 
     # The job in the background queue would be failed by the supervisor
     # when it replaced the killed worker

test/models/solid_queue/claimed_execution_concurrency_test.rb

@@ -0,0 +1,231 @@
+# frozen_string_literal: true
+
+require "test_helper"
+
+# These tests exercise the real FOR UPDATE serialization in ClaimedExecution's
+# finalization, so they run on separate connections/threads and skip on SQLite,
+# which has no row-level locking. They follow the same pattern as
+# SolidQueue::SemaphoreTest.
+class SolidQueue::ClaimedExecutionConcurrencyTest < ActiveSupport::TestCase
+  self.use_transactional_tests = false
+
+  setup do
+    @process = SolidQueue::Process.register(kind: "Worker", pid: 42, name: "worker-123")
+  end
+
+  test "a stale performer that loses the claim lock to pruning does not finish or re-promote a pruned job" do
+    skip_on_sqlite
+
+    first_job, claimed_execution, concurrency_key = enqueue_blocked_group
+
+    elapsed = with_winner_holding_claim(claimed_execution,
+      in_transaction: ->(locked_claim) do
+        first_job.failed_with(SolidQueue::Processes::ProcessPrunedError.new(1.day.ago))
+        locked_claim.destroy!
+        first_job.release_concurrency_permit
+      end,
+      after_commit: -> { first_job.promote_next_blocked_job }) do
+      claimed_execution.perform
+    end
+
+    assert_operator elapsed, :>=, 0.3, "performer should have blocked on the claim lock (took #{elapsed.round(3)}s)"
+    assert first_job.reload.failed?
+    assert_not first_job.finished?
+    assert_equal 1, SolidQueue::ReadyExecution.count
+    assert_equal 1, SolidQueue::BlockedExecution.count
+    assert_equal 0, SolidQueue::ClaimedExecution.count
+    assert_equal 0, SolidQueue::Semaphore.find_by!(key: concurrency_key).value
+  end
+
+  test "a stale failing performer that loses the claim lock to pruning does not double-finalize" do
+    skip_on_sqlite
+
+    first_job, claimed_execution, concurrency_key = enqueue_blocked_group(exception: RuntimeError)
+
+    elapsed = with_winner_holding_claim(claimed_execution,
+      in_transaction: ->(locked_claim) do
+        first_job.failed_with(SolidQueue::Processes::ProcessPrunedError.new(1.day.ago))
+        locked_claim.destroy!
+        first_job.release_concurrency_permit
+      end,
+      after_commit: -> { first_job.promote_next_blocked_job }) do
+      assert_raises(RuntimeError) { claimed_execution.perform }
+    end
+
+    assert_operator elapsed, :>=, 0.3, "performer should have blocked on the claim lock (took #{elapsed.round(3)}s)"
+    assert first_job.reload.failed?
+    assert_equal 1, SolidQueue::ReadyExecution.count
+    assert_equal 1, SolidQueue::BlockedExecution.count
+    assert_equal 1, SolidQueue::FailedExecution.count
+    assert_equal 0, SolidQueue::ClaimedExecution.count
+    assert_equal 0, SolidQueue::Semaphore.find_by!(key: concurrency_key).value
+  end
+
+  test "a stale release that loses the claim lock to pruning does not re-dispatch a pruned job" do
+    skip_on_sqlite
+
+    first_job, claimed_execution, concurrency_key = enqueue_blocked_group
+
+    elapsed = with_winner_holding_claim(claimed_execution,
+      in_transaction: ->(locked_claim) do
+        first_job.failed_with(SolidQueue::Processes::ProcessPrunedError.new(1.day.ago))
+        locked_claim.destroy!
+        first_job.release_concurrency_permit
+      end,
+      after_commit: -> { first_job.promote_next_blocked_job }) do
+      claimed_execution.release
+    end
+
+    assert_operator elapsed, :>=, 0.3, "release should have blocked on the claim lock (took #{elapsed.round(3)}s)"
+    assert first_job.reload.failed?
+    assert_not first_job.ready?
+    assert_equal 1, SolidQueue::ReadyExecution.count
+    assert_equal 1, SolidQueue::BlockedExecution.count
+    assert_equal 0, SolidQueue::ClaimedExecution.count
+    assert_equal 0, SolidQueue::Semaphore.find_by!(key: concurrency_key).value
+  end
+
+  test "pruning that loses the claim lock to a finishing performer does not fail a finished job" do
+    skip_on_sqlite
+
+    first_job, claimed_execution, concurrency_key = enqueue_blocked_group
+
+    elapsed = with_winner_holding_claim(claimed_execution,
+      in_transaction: ->(locked_claim) do
+        first_job.finished!
+        locked_claim.destroy!
+        first_job.release_concurrency_permit
+      end,
+      after_commit: -> { first_job.promote_next_blocked_job }) do
+      @process.prune
+    end
+
+    assert_operator elapsed, :>=, 0.3, "pruning should have blocked on the claim lock (took #{elapsed.round(3)}s)"
+    assert first_job.reload.finished?
+    assert_not first_job.failed?
+    assert_equal 1, SolidQueue::ReadyExecution.count
+    assert_equal 1, SolidQueue::BlockedExecution.count
+    assert_equal 0, SolidQueue::FailedExecution.count
+    assert_equal 0, SolidQueue::ClaimedExecution.count
+    assert_equal 0, SolidQueue::Semaphore.find_by!(key: concurrency_key).value
+  end
+
+  test "finalizing a claim does not deadlock a concurrent blocked-job release when skip_locked is off" do
+    skip_on_sqlite
+
+    with_skip_locked(false) do
+      first_job, claimed_execution, concurrency_key = enqueue_blocked_group
+      blocked_execution = SolidQueue::BlockedExecution.first
+
+      holding_blocked_row = Concurrent::Event.new
+      release_blocked_row = Concurrent::Event.new
+
+      # Mimic the normal release path's lock order: lock the blocked row first,
+      # then contend for the semaphore. Held open so finalize must run alongside it.
+      releaser = Thread.new do
+        SolidQueue::Record.connection_pool.with_connection do
+          SolidQueue::Record.transaction do
+            SolidQueue::BlockedExecution.where(id: blocked_execution.id).lock.first
+            holding_blocked_row.set
+            release_blocked_row.wait(5)
+            SolidQueue::Semaphore.where(key: concurrency_key).lock.first
+          end
+        end
+      end
+
+      holding_blocked_row.wait(5)
+
+      # finalize returns the permit and commits before it tries to promote, so it
+      # never holds the semaphore lock while waiting on the blocked row. Under the
+      # old in-transaction promotion this reverse lock order could deadlock.
+      assert_nothing_raised do
+        Timeout.timeout(10) do
+          finisher = Thread.new do
+            SolidQueue::Record.connection_pool.with_connection do
+              claimed_execution.send(:finished)
+            end
+          end
+
+          # Give finalize time to commit its semaphore signal, then let the
+          # releaser reach for the semaphore.
+          sleep 0.5
+          release_blocked_row.set
+          finisher.join(10)
+        end
+      end
+
+      releaser.join(10)
+
+      assert first_job.reload.finished?
+      assert_equal 0, SolidQueue::ClaimedExecution.count
+    end
+  end
+
+  private
+    # Enqueues three concurrency-limited jobs over the same key. The first is
+    # claimed; the other two are blocked. Returns the first job, its claimed
+    # execution, and the shared concurrency key.
+    def enqueue_blocked_group(exception: nil)
+      job_result = JobResult.create!(queue_name: "default", status: "")
+      first_active_job = NonOverlappingUpdateResultJob.perform_later(job_result, name: "A", exception: exception)
+      NonOverlappingUpdateResultJob.perform_later(job_result, name: "B")
+      NonOverlappingUpdateResultJob.perform_later(job_result, name: "C")
+
+      first_job = SolidQueue::Job.find_by!(active_job_id: first_active_job.job_id)
+      claimed_execution = claim_first_job(first_job)
+
+      [ first_job, claimed_execution, first_job.concurrency_key ]
+    end
+
+    def claim_first_job(job)
+      SolidQueue::ReadyExecution.claim(job.queue_name, 1, @process.id)
+      SolidQueue::ClaimedExecution.find_by!(job_id: job.id)
+    end
+
+    # Runs the given block (the stale worker's action) while a separate connection
+    # holds the claim's FOR UPDATE lock, then finishes the claim first and commits.
+    # The stale worker must block on the lock and, once it resumes, find the claim
+    # gone. Returns how long the stale worker was blocked.
+    def with_winner_holding_claim(claimed_execution, in_transaction:, after_commit: nil)
+      claim_locked = Concurrent::Event.new
+
+      winner = Thread.new do
+        SolidQueue::Record.connection_pool.with_connection do
+          SolidQueue::Record.transaction do
+            locked_claim = SolidQueue::ClaimedExecution.unscoped.lock.find_by(id: claimed_execution.id)
+            claim_locked.set
+            sleep 0.5
+            in_transaction.call(locked_claim)
+          end
+
+          after_commit&.call
+        end
+      end
+
+      claim_locked.wait(5)
+      sleep 0.1
+
+      started_at = monotonic_now
+      yield
+      elapsed = monotonic_now - started_at
+
+      winner.join(5)
+      elapsed
+    end
+
+    def with_skip_locked(value)
+      previous = SolidQueue.use_skip_locked
+      SolidQueue.use_skip_locked = value
+      yield
+    ensure
+      SolidQueue.use_skip_locked = previous
+    end
+
+    def skip_on_sqlite
+      skip "Row-level locking not supported on SQLite" if SolidQueue::Record.connection.adapter_name.downcase.include?("sqlite")
+    end
+
+    def monotonic_now
+      Process.clock_gettime(Process::CLOCK_MONOTONIC)
+    end
+end