From 9c01041f217f7a66f84dfc4fb5fb1e504d7f8269 Mon Sep 17 00:00:00 2001 From: Tom Blais Date: Fri, 10 Jul 2026 15:21:17 -0400 Subject: [PATCH 1/4] Prevent false pruning from leaking concurrency permits * Keep stale supervised workers while their supervisor heartbeat is healthy * Make claimed execution finalization own and release permits exactly once * Add regression coverage and document pruning behavior --- README.md | 2 +- app/models/solid_queue/claimed_execution.rb | 29 ++++++++++++------- app/models/solid_queue/process/prunable.rb | 9 +++++- .../solid_queue/claimed_execution_test.rb | 23 +++++++++++++++ test/models/solid_queue/process_test.rb | 22 ++++++++++++++ 5 files changed, 72 insertions(+), 13 deletions(-) diff --git a/README.md b/README.md index 629d0d7c2..589b40a1b 100644 --- a/README.md +++ b/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. diff --git a/app/models/solid_queue/claimed_execution.rb b/app/models/solid_queue/claimed_execution.rb index 5d0a4057f..3e9477ca3 100644 --- a/app/models/solid_queue/claimed_execution.rb +++ b/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 diff --git a/app/models/solid_queue/process/prunable.rb b/app/models/solid_queue/process/prunable.rb index 85341d1d7..c85b433d8 100644 --- a/app/models/solid_queue/process/prunable.rb +++ b/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 diff --git a/test/models/solid_queue/claimed_execution_test.rb b/test/models/solid_queue/claimed_execution_test.rb index dd1088c90..148eef42d 100644 --- a/test/models/solid_queue/claimed_execution_test.rb +++ b/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 diff --git a/test/models/solid_queue/process_test.rb b/test/models/solid_queue/process_test.rb index cd2430caf..947805e26 100644 --- a/test/models/solid_queue/process_test.rb +++ b/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" From 927ac7d38f2cdbf44d9d848688268c11ae5a8fac Mon Sep 17 00:00:00 2001 From: Tom Blais Date: Tue, 14 Jul 2026 10:25:53 -0400 Subject: [PATCH 2/4] Promote blocked jobs after claim finalization commits * Split unblock_next_blocked_job into release_concurrency_permit and promote_next_blocked_job so promotion runs after the claim transaction commits, matching the failure isolation from 873760d * Returning the permit inside the claim transaction keeps ownership atomic; promoting outside it means a promotion failure can no longer roll back a job that already finished or failed, and avoids holding the semaphore lock while waiting on the blocked row's lock --- app/models/solid_queue/claimed_execution.rb | 10 +++++++++- .../solid_queue/job/concurrency_controls.rb | 15 ++++++++++++--- 2 files changed, 21 insertions(+), 4 deletions(-) diff --git a/app/models/solid_queue/claimed_execution.rb b/app/models/solid_queue/claimed_execution.rb index 3e9477ca3..f7ee781f9 100644 --- a/app/models/solid_queue/claimed_execution.rb +++ b/app/models/solid_queue/claimed_execution.rb @@ -106,11 +106,19 @@ def finished end def finalize_claim + permit_released = false + with_claim do yield destroy! - job.unblock_next_blocked_job + # 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 diff --git a/app/models/solid_queue/job/concurrency_controls.rb b/app/models/solid_queue/job/concurrency_controls.rb index 86394ef56..2c8a809e0 100644 --- a/app/models/solid_queue/job/concurrency_controls.rb +++ b/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? From 244eafe09d16bebdda8f77cadadda2af7fa172c4 Mon Sep 17 00:00:00 2001 From: Tom Blais Date: Tue, 14 Jul 2026 10:26:10 -0400 Subject: [PATCH 3/4] Deregister reaped processes and add claim-lock race coverage * Deregister a supervised process's row as soon as its supervisor reaps it, mirroring Process::Prunable#prune, so a dead child no longer stays permanently protected from pruning just because its supervisor is alive * Update the two existing tests whose expected process counts depended on that dead row lingering * Add non-transactional, multi-connection tests that exercise the actual FOR UPDATE serialization in ClaimedExecution finalization: both race outcomes (pruner wins, performer wins) across finish/fail/release paths, and a use_skip_locked=false regression guarding against the reverse lock order between the semaphore and the blocked row --- lib/solid_queue/supervisor/maintenance.rb | 1 + .../forked_processes_lifecycle_test.rb | 10 +- .../claimed_execution_concurrency_test.rb | 231 ++++++++++++++++++ test/unit/process_recovery_test.rb | 6 +- 4 files changed, 244 insertions(+), 4 deletions(-) create mode 100644 test/models/solid_queue/claimed_execution_concurrency_test.rb diff --git a/lib/solid_queue/supervisor/maintenance.rb b/lib/solid_queue/supervisor/maintenance.rb index d92569d58..f21e03098 100644 --- a/lib/solid_queue/supervisor/maintenance.rb +++ b/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 diff --git a/test/integration/forked_processes_lifecycle_test.rb b/test/integration/forked_processes_lifecycle_test.rb index 40495c20a..14d50be0c 100644 --- a/test/integration/forked_processes_lifecycle_test.rb +++ b/test/integration/forked_processes_lifecycle_test.rb @@ -244,8 +244,14 @@ 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. A plain count of 3 can't distinguish "not reaped yet" from + # "reaped and replaced" (both are 3), so wait on the dead row disappearing. + wait_while_with_timeout(5.seconds) { SolidQueue::Process.exists?(id: worker.id) } + 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 diff --git a/test/models/solid_queue/claimed_execution_concurrency_test.rb b/test/models/solid_queue/claimed_execution_concurrency_test.rb new file mode 100644 index 000000000..b48b68376 --- /dev/null +++ b/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 diff --git a/test/unit/process_recovery_test.rb b/test/unit/process_recovery_test.rb index 296d6b958..03ed7fbbf 100644 --- a/test/unit/process_recovery_test.rb +++ b/test/unit/process_recovery_test.rb @@ -51,8 +51,10 @@ class ProcessRecoveryTest < ActiveSupport::TestCase assert failed_execution.present? assert_equal "SolidQueue::Processes::ProcessExitError", failed_execution.exception_class - # Ensure supervisor replaces the worker (even though its own record was missing) - wait_for_registered_processes(2, timeout: 5.seconds) + # Ensure supervisor replaces the worker (even though its own record was missing). + # The killed worker's row is deregistered on reap, and the supervisor's own + # record was deleted above, so only the replacement worker remains registered. + wait_for_registered_processes(1, timeout: 5.seconds) assert_operator SolidQueue::Process.where(kind: "Worker").count, :>=, 1 end end From 715fc74a25932f0aa12f3b43e120ed8863f39e35 Mon Sep 17 00:00:00 2001 From: Brice Burgess Date: Thu, 16 Jul 2026 14:39:09 -0600 Subject: [PATCH 4/4] Fix flaky process-count assertion in kill-worker lifecycle test test_kill_worker_individually waited only for the killed worker's row to be reaped, then asserted Process.count == 3. The reap and the replacement worker's registration are separate async events, so CI could observe the transient window where the dead row was gone but the replacement had not registered yet (count 2), failing deterministically across the matrix. Wait for the fleet to return to full strength before asserting. --- test/integration/forked_processes_lifecycle_test.rb | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/test/integration/forked_processes_lifecycle_test.rb b/test/integration/forked_processes_lifecycle_test.rb index 14d50be0c..f104d019b 100644 --- a/test/integration/forked_processes_lifecycle_test.rb +++ b/test/integration/forked_processes_lifecycle_test.rb @@ -245,9 +245,12 @@ class ForkedProcessesLifecycleTest < ActiveSupport::TestCase assert SolidQueue::Process.exists?(id: worker.id) # The supervisor reaps the killed worker, deregisters it, and starts a - # replacement. A plain count of 3 can't distinguish "not reaped yet" from - # "reaped and replaced" (both are 3), so wait on the dead row disappearing. + # 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