Fix flaky tests caused by leaked job threads and signal timing

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)

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
 

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)

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

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

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)