diff --git a/test/integration/async_processes_lifecycle_test.rb b/test/integration/async_processes_lifecycle_test.rb index 6f269e220..3ccd4d461 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/concurrency_controls_test.rb b/test/integration/concurrency_controls_test.rb index 01fdf8a11..9cdb3d65d 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/forked_processes_lifecycle_test.rb b/test/integration/forked_processes_lifecycle_test.rb index 40ade6910..deb196d31 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) diff --git a/test/integration/instrumentation_test.rb b/test/integration/instrumentation_test.rb index 1822cf159..88f3fbaff 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 a776f703d..7e9c1568f 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 01cec796c..4a521b568 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)