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,24 @@ def execute
     end
 
     def finished
-      transaction do
+      finalize_claim do
         job.finished!
+      end
+    end
+
+    def finalize_claim
+      with_claim do
+        yield
         destroy!
+        job.unblock_next_blocked_job
+      end
+    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/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

test/models/solid_queue/claimed_execution_test.rb

@@ -17,6 +17,29 @@ class SolidQueue::ClaimedExecutionTest < ActiveSupport::TestCase
     assert job.reload.finished?
   end
 
+  test "stale performer cannot return a concurrency permit after its claim is pruned" do
+    job_result = JobResult.create!(queue_name: "default", status: "")
+    first_active_job = NonOverlappingUpdateResultJob.perform_later(job_result, name: "A")
+    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_job(first_job)
+    @process.prune
+
+    assert first_job.reload.failed?
+    assert_equal 1, SolidQueue::ReadyExecution.count
+    assert_equal 1, SolidQueue::BlockedExecution.count
+
+    claimed_execution.perform
+
+    assert_not first_job.reload.finished?
+    assert first_job.failed?
+    assert_equal 1, SolidQueue::ReadyExecution.count
+    assert_equal 1, SolidQueue::BlockedExecution.count
+    assert_equal 0, SolidQueue::Semaphore.find_by!(key: first_job.concurrency_key).value
+  end
+
   test "perform job that fails" do
     claimed_execution = prepare_and_claim_job RaisingJob.perform_later(RuntimeError, "A")
     job = claimed_execution.job

test/models/solid_queue/process_test.rb

@@ -55,6 +55,28 @@ class SolidQueue::ProcessTest < ActiveSupport::TestCase
     assert jobs.all?(&:failed?)
   end
 
+  test "keep a stale supervised process while its supervisor is alive" do
+    supervisor = SolidQueue::Process.register(
+      kind: "Supervisor(fork)", pid: 42, name: "supervisor-42"
+    )
+    worker = SolidQueue::Process.register(
+      kind: "Worker", pid: 43, name: "worker-43", supervisor_id: supervisor.id
+    )
+    StoreResultJob.set(queue: :new_queue).perform_later(42)
+    SolidQueue::ReadyExecution.claim("*", 1, worker.id)
+
+    travel_to 10.minutes.from_now
+    supervisor.heartbeat
+
+    assert_no_difference -> { SolidQueue::Process.count } do
+      assert_no_difference -> { SolidQueue::ClaimedExecution.count } do
+        assert_no_difference -> { SolidQueue::FailedExecution.count } do
+          SolidQueue::Process.prune
+        end
+      end
+    end
+  end
+
   test "hostname's with special characters are properly loaded" do
     worker = SolidQueue::Worker.new(queues: "*", threads: 3, polling_interval: 0.2)
     hostname = "Basecamp's-Computer"