Reduce flaky integration test failures from timing and PK reuse

test/integration/async_processes_lifecycle_test.rb

@@ -58,7 +58,7 @@ class AsyncProcessesLifecycleTest < ActiveSupport::TestCase
       signal_process(@pid, :TERM, wait: 0.1.second)
     end
 
-    sleep(1.second)
+    wait_while_with_timeout(SolidQueue.shutdown_timeout + 1.second) { process_exists?(@pid) }
     assert_clean_termination
   end
 

test/integration/concurrency_controls_test.rb

@@ -6,18 +6,27 @@ class ConcurrencyControlsTest < ActiveSupport::TestCase
   self.use_transactional_tests = false
 
   setup do
-    @result = JobResult.create!(queue_name: "default", status: "")
+    # 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_registered_processes(0, timeout: 5.seconds)
+    destroy_records
 
     default_worker = { queues: "default", polling_interval: 0.1, processes: 3, threads: 2 }
     dispatcher = { polling_interval: 0.1, batch_size: 200, concurrency_maintenance_interval: 1 }
 
     @pid = run_supervisor_as_fork(workers: [ default_worker ], dispatchers: [ dispatcher ])
+    wait_for_registered_processes(5, timeout: 3.seconds) # 3 workers + dispatcher + supervisor
 
-    wait_for_registered_processes(5, timeout: 0.5.second) # 3 workers working the default queue + dispatcher + supervisor
+    @result = JobResult.create!(queue_name: "default", status: "")
   end
 
   teardown do
-    terminate_process(@pid) if process_exists?(@pid)
+    if @pid && process_exists?(@pid)
+      terminate_process(@pid)
+    end
+    wait_for_registered_processes(0, timeout: 5.seconds)
+    destroy_records
   end
 
   test "run several conflicting jobs over the same record without overlapping" do
@@ -48,7 +57,7 @@ class ConcurrencyControlsTest < ActiveSupport::TestCase
       NonOverlappingUpdateResultJob.set(wait: (1 + i * 0.1).seconds).perform_later(@result, name: name)
     end
 
-    wait_for_jobs_to_finish_for(5.seconds)
+    wait_for_jobs_to_finish_for(15.seconds)
     assert_no_unfinished_jobs
 
     assert_stored_sequence @result, ("A".."K").to_a
@@ -67,16 +76,18 @@ class ConcurrencyControlsTest < ActiveSupport::TestCase
       end
     end
 
-    sleep(0.01) # To ensure these aren't picked up before ABC
-    # D to H: 0.51 to 0.76 (starting after A finishes, and in order, 5 * 0.05 = 0.25)
+    # Wait until A/B/C hold the concurrency slots so D–H are blocked instead of claimed
+    wait_for(timeout: 2.seconds) { SolidQueue::ClaimedExecution.count >= 3 }
+
+    # D to H: starting after A finishes, and in order, 5 * 0.05 = 0.25
     # These would finish all before B and C
     assert_difference -> { SolidQueue::BlockedExecution.count }, +5 do
       ("D".."H").each do |name|
         ThrottledUpdateResultJob.perform_later(@result, name: name, pause: 0.05.seconds)
       end
     end
 
-    wait_for_jobs_to_finish_for(5.seconds)
+    wait_for_jobs_to_finish_for(15.seconds)
     assert_no_unfinished_jobs
 
     # C would have started in the beginning, seeing the status empty, and would finish after
@@ -94,7 +105,7 @@ class ConcurrencyControlsTest < ActiveSupport::TestCase
       NonOverlappingUpdateResultJob.perform_later(@result, name: name)
     end
 
-    wait_for_jobs_to_finish_for(5.seconds)
+    wait_for_jobs_to_finish_for(15.seconds)
     assert_equal 3, SolidQueue::FailedExecution.count
 
     assert_stored_sequence @result, [ "B", "D", "F" ] + ("G".."K").to_a
@@ -166,12 +177,11 @@ class ConcurrencyControlsTest < ActiveSupport::TestCase
     NonOverlappingUpdateResultJob.perform_later(@result, name: "I'll be released to ready", pause: SolidQueue.shutdown_timeout + 10.seconds)
     job = SolidQueue::Job.last
 
-    sleep(0.2)
-    assert job.claimed?
+    wait_for(timeout: 2.seconds) { job.reload.claimed? }
 
-    # This won't leave time to the job to finish
-    signal_process(@pid, :TERM, wait: 0.1.second)
-    sleep(SolidQueue.shutdown_timeout + 0.6.seconds)
+    # This won't leave time to the job to finish, so the worker should
+    # release it back to ready during shutdown.
+    terminate_process(@pid)
 
     assert_not job.reload.finished?
     assert job.reload.ready?
@@ -196,8 +206,8 @@ class ConcurrencyControlsTest < ActiveSupport::TestCase
   end
 
   test "discard jobs when concurrency limit is reached with on_conflict: :discard" do
-    job1 = DiscardableUpdateResultJob.perform_later(@result, name: "1", pause: 3)
-    sleep(0.1)
+    job1 = DiscardableUpdateResultJob.perform_later(@result, name: "1", pause: 1.second)
+    wait_for(timeout: 2.seconds) { SolidQueue::Job.find_by(active_job_id: job1.job_id)&.claimed? }
 
     # should be discarded due to concurrency limit
     job2 = DiscardableUpdateResultJob.perform_later(@result, name: "2")

test/integration/jobs_lifecycle_test.rb

@@ -40,11 +40,15 @@ class JobsLifecycleTest < ActiveSupport::TestCase
     @dispatcher.start
     @worker.start
 
-    wait_for_jobs_to_finish_for(3.seconds)
+    wait_while_with_timeout(3.seconds) { SolidQueue::FailedExecution.count < 2 }
+
+    @worker.stop
+    @dispatcher.stop
 
     message = "raised ExpectedTestError for the 1st time"
     assert_equal [ "A: #{message}", "B: #{message}" ], JobBuffer.values.sort
 
+    assert_equal 2, SolidQueue::FailedExecution.count
     assert_empty SolidQueue::Job.finished
   end
 

test/integration/recurring_tasks_test.rb

@@ -24,11 +24,10 @@ class RecurringTasksTest < ActiveSupport::TestCase
         assert_equal "StoreResultJob", job.class_name
       end
 
-      assert JobResult.count >= 2
-      JobResult.all.each do |result|
-        assert_equal "custom_status", result.status
-        assert_equal "42", result.value
-      end
+      # Scope to this recurring task's results. Other tests may leave JobResult
+      # rows (e.g. status "completed") that would fail a JobResult.all assertion.
+      results = JobResult.where(status: "custom_status", value: "42")
+      assert results.count >= 2
     end
   end