From c8cf6c89743269ebb95d5fe93e82adb4fbc5dc8e Mon Sep 17 00:00:00 2001 From: Rosa Gutierrez Date: Fri, 24 Jul 2026 16:07:50 +0200 Subject: [PATCH 1/2] Wait deterministically before signaling QUIT and KILL in lifecycle tests The tests only waited for jobs to be claimed before signaling, so on a slow runner the signal's exit! could land after the instant job was claimed, or even after it had written its result, but before its finished_at was committed, leaving it unfinished and failing the test. Wait until the instant job has actually finished and the paused job has actually started before signaling, so the signal always arrives while only the paused job is in-flight, which is the scenario these tests exist to check. The async QUIT test's pause also becomes 60 seconds like the forked variant's: with only 1 second, a slow QUIT could land after the paused job finished, breaking the assertion that it remains claimed. Nothing waits for the pause, so test runtime is unaffected. Co-Authored-By: Claude Opus 4.8 --- .../integration/async_processes_lifecycle_test.rb | 15 ++++++++++----- .../forked_processes_lifecycle_test.rb | 12 +++++++----- 2 files changed, 17 insertions(+), 10 deletions(-) diff --git a/test/integration/async_processes_lifecycle_test.rb b/test/integration/async_processes_lifecycle_test.rb index 6f269e22..3ccd4d46 100644 --- a/test/integration/async_processes_lifecycle_test.rb +++ b/test/integration/async_processes_lifecycle_test.rb @@ -34,8 +34,10 @@ class AsyncProcessesLifecycleTest < ActiveSupport::TestCase no_pause = enqueue_store_result_job("no pause") pause = enqueue_store_result_job("pause", pause: 3.second) - # Wait for the "no pause" job to complete before sending KILL + # Wait for the "no pause" job to complete and the "pause" job to start + # before sending KILL, so it always arrives while "pause" is in-flight wait_for_jobs_to_finish_for(2.seconds, except: pause) + wait_while_with_timeout(3.seconds) { !JobResult.exists?(status: "started", value: "pause") } signal_process(@pid, :KILL, wait: 0.1.seconds) wait_for_registered_processes(1, timeout: 2.second) @@ -64,12 +66,15 @@ class AsyncProcessesLifecycleTest < ActiveSupport::TestCase test "quit supervisor while there are jobs in-flight" do no_pause = enqueue_store_result_job("no pause") - pause = enqueue_store_result_job("pause", pause: 1.second) + # long enough pause to make sure it doesn't finish before QUIT arrives + pause = enqueue_store_result_job("pause", pause: 60.seconds) - wait_while_with_timeout(1.second) { SolidQueue::ReadyExecution.count > 0 } + # Wait for the "no pause" job to finish and the "pause" job to start + # before sending QUIT, so it always arrives while "pause" is in-flight + wait_for_jobs_to_finish_for(3.seconds, except: pause) + wait_while_with_timeout(3.seconds) { !JobResult.exists?(status: "started", value: "pause") } - signal_process(@pid, :QUIT, wait: 0.4.second) - wait_for_jobs_to_finish_for(2.seconds, except: pause) + signal_process(@pid, :QUIT, wait: 0.1.second) wait_while_with_timeout(2.seconds) { process_exists?(@pid) } assert_not process_exists?(@pid) diff --git a/test/integration/forked_processes_lifecycle_test.rb b/test/integration/forked_processes_lifecycle_test.rb index 40ade691..deb196d3 100644 --- a/test/integration/forked_processes_lifecycle_test.rb +++ b/test/integration/forked_processes_lifecycle_test.rb @@ -64,13 +64,15 @@ class ForkedProcessesLifecycleTest < ActiveSupport::TestCase test "quit supervisor while there are jobs in-flight" do no_pause = enqueue_store_result_job("no pause") - # long enough pause to make sure it doesn't finish - pause = enqueue_store_result_job("pause", pause: 60.second) + # long enough pause to make sure it doesn't finish before QUIT arrives + pause = enqueue_store_result_job("pause", pause: 60.seconds) - wait_while_with_timeout(1.second) { SolidQueue::ReadyExecution.count > 0 } + # Wait for the "no pause" job to finish and the "pause" job to start + # before sending QUIT, so it always arrives while "pause" is in-flight + wait_for_jobs_to_finish_for(3.seconds, except: pause) + wait_while_with_timeout(3.seconds) { !JobResult.exists?(status: "started", value: "pause") } - signal_process(@pid, :QUIT, wait: 0.4.second) - wait_for_jobs_to_finish_for(2.seconds, except: pause) + signal_process(@pid, :QUIT, wait: 0.1.second) wait_while_with_timeout(2.seconds) { process_exists?(@pid) } assert_not process_exists?(@pid) From c6aeb9c11822bb37a7fa2a926ef2b5e9b8a3d34f Mon Sep 17 00:00:00 2001 From: Rosa Gutierrez Date: Fri, 24 Jul 2026 16:07:50 +0200 Subject: [PATCH 2/2] Kill job threads leaked by workers stopped in-process in tests A worker stopped while a job is running gives up on its pool after SolidQueue.shutdown_timeout and releases the claim, but it can't stop the thread running the job. A forked worker exits right after, killing the thread; a worker running in-process in a test leaks it instead, sleeping until the job's pause is over, long after its test finished. Inside a transactional test this corrupts later tests: the leaked job's inserts roll back with the test, and on SQLite the AUTOINCREMENT sequence bump rolls back too, so later tests' rows get the same ids and the leaked thread writes over them when it wakes up. This was behind several flaky failures on SQLite: a job result overwritten with "completed" mid-assertion, and a claimed execution stolen by a leaked finalize call on a recycled claim id, leaving a job neither finished nor ready. It can't happen in production: ids are only re-issued when the insert rolls back, and a real worker's threads die with its process. Kill the leaked threads in the tests that create them, like a forked worker's exit would, and fix the explanation of SQLite id reuse in the concurrency controls test setup: with AUTOINCREMENT, committed deletes never free ids; it's the transactional rollback that recycles them. Co-Authored-By: Claude Opus 4.8 --- test/integration/concurrency_controls_test.rb | 6 +++--- test/integration/instrumentation_test.rb | 6 ++++++ test/models/solid_queue/job_test.rb | 2 ++ test/test_helpers/processes_test_helper.rb | 14 ++++++++++++++ 4 files changed, 25 insertions(+), 3 deletions(-) diff --git a/test/integration/concurrency_controls_test.rb b/test/integration/concurrency_controls_test.rb index 01fdf8a1..9cdb3d65 100644 --- a/test/integration/concurrency_controls_test.rb +++ b/test/integration/concurrency_controls_test.rb @@ -6,9 +6,9 @@ class ConcurrencyControlsTest < ActiveSupport::TestCase self.use_transactional_tests = false setup do - # Previous tests may leave forked workers briefly alive; those can still write to - # JobResult rows whose primary keys get reused by create! below (e.g. overwriting - # status with StoreResultJob's default "completed"). + # Wait for processes from previous tests to deregister first: leftover workers + # polling the same queues could claim this test's jobs. Then clear any records + # they might have left behind. wait_for_registered_processes(0, timeout: 5.seconds) destroy_records diff --git a/test/integration/instrumentation_test.rb b/test/integration/instrumentation_test.rb index 1822cf15..88f3fbaf 100644 --- a/test/integration/instrumentation_test.rb +++ b/test/integration/instrumentation_test.rb @@ -55,6 +55,7 @@ class InstrumentationTest < ActiveSupport::TestCase test "stopping a worker with claimed executions emits release_claimed events" do StoreResultJob.perform_later(42, pause: SolidQueue.shutdown_timeout + 15.seconds) process = nil + worker = nil events = subscribed(/release.*_claimed\.solid_queue/) do worker = SolidQueue::Worker.new.tap(&:start) @@ -70,6 +71,8 @@ class InstrumentationTest < ActiveSupport::TestCase release_one_event, release_many_event = events assert_event release_one_event, "release_claimed", job_id: SolidQueue::Job.last.id, process_id: process.id assert_event release_many_event, "release_many_claimed", size: 1 + ensure + kill_running_jobs_in(worker) end test "starting a runnable process emits a start_process event" do @@ -90,6 +93,7 @@ class InstrumentationTest < ActiveSupport::TestCase test "starting and stopping a worker emits register_process and deregister_process events" do StoreResultJob.perform_later(42, pause: SolidQueue.shutdown_timeout + 15.seconds) process = nil + worker = nil events = subscribed(/(register|deregister)_process\.solid_queue/) do worker = SolidQueue::Worker.new.tap(&:start) @@ -105,6 +109,8 @@ class InstrumentationTest < ActiveSupport::TestCase register_event, deregister_event = events assert_event register_event, "register_process", kind: "Worker", pid: ::Process.pid, process_id: process.id assert_event deregister_event, "deregister_process", process: process, pruned: false + ensure + kill_running_jobs_in(worker) end test "starting and stopping a dispatcher emits register_process and deregister_process events" do diff --git a/test/models/solid_queue/job_test.rb b/test/models/solid_queue/job_test.rb index a776f703..7e9c1568 100644 --- a/test/models/solid_queue/job_test.rb +++ b/test/models/solid_queue/job_test.rb @@ -265,6 +265,8 @@ class DiscardableNonOverlappingGroupedJob2 < NonOverlappingJob end worker.stop + ensure + kill_running_jobs_in(worker) end test "discard scheduled job" do diff --git a/test/test_helpers/processes_test_helper.rb b/test/test_helpers/processes_test_helper.rb index 01cec796..4a521b56 100644 --- a/test/test_helpers/processes_test_helper.rb +++ b/test/test_helpers/processes_test_helper.rb @@ -65,6 +65,20 @@ def terminate_process(pid, timeout: 10, signal: :TERM) wait_for_process_termination_with_timeout(pid, timeout: timeout, signaled: signal) end + # A worker stopped while a job is still running gives up on its pool after + # SolidQueue.shutdown_timeout and releases the claim, but it can't stop the + # thread that is running the job. A forked worker exits right after, killing + # the thread; a worker running in-process in a test leaks it instead, and the + # thread writes to the database whenever the job's pause is over — long after + # the test that started it has finished. Inside a transactional test this + # corrupts later tests: the job's inserts roll back with the test, on SQLite + # rolling back the AUTOINCREMENT sequence bump too, so a later test's rows + # get the same ids and the leaked thread overwrites them when it wakes up. + # Kill the leaked threads instead, like a forked worker's exit would. + def kill_running_jobs_in(worker) + worker&.pool&.send(:executor)&.kill + end + def wait_for_process_termination_with_timeout(pid, timeout: 10, exitstatus: 0, signaled: nil) Timeout.timeout(timeout) do if process_exists?(pid)