Reduce flaky integration test failures from timing and PK reuse

test/integration/async_processes_lifecycle_test.rb

@@ -22,7 +22,7 @@ class AsyncProcessesLifecycleTest < ActiveSupport::TestCase
 
     wait_for_jobs_to_finish_for(2.seconds)
 
-    assert_equal 12, JobResult.count
+    assert_equal 12, skip_active_record_query_cache { JobResult.count }
     6.times { |i| assert_completed_job_results("job_#{i}", :background) }
     6.times { |i| assert_completed_job_results("job_#{i}", :default) }
 
@@ -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
 
@@ -212,15 +212,17 @@ def enqueue_store_result_job(value, queue_name = :background, **options)
     end
 
     def assert_completed_job_results(value, queue_name = :background, count = 1)
-      skip_active_record_query_cache do
-        assert_equal count, JobResult.where(queue_name: queue_name, status: "completed", value: value).count
-      end
+      actual = skip_active_record_query_cache {
+        JobResult.where(queue_name: queue_name, status: "completed", value: value).count
+      }
+      assert_equal count, actual
     end
 
     def assert_started_job_result(value, queue_name = :background, count = 1)
-      skip_active_record_query_cache do
-        assert_equal count, JobResult.where(queue_name: queue_name, status: "started", value: value).count
-      end
+      actual = skip_active_record_query_cache {
+        JobResult.where(queue_name: queue_name, status: "started", value: value).count
+      }
+      assert_equal count, actual
     end
 
     def assert_job_status(active_job, status)

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
@@ -36,51 +45,53 @@ class ConcurrencyControlsTest < ActiveSupport::TestCase
   end
 
   test "schedule several conflicting jobs over the same record sequentially" do
-    # Writes to @result at 0.4s
-    UpdateResultJob.set(wait: 0.2.seconds).perform_later(@result, name: "000", pause: 0.2.seconds)
-
-    ("A".."F").each_with_index do |name, i|
-      # "A" is enqueued at 0.2s and writes to @result at 0.6s, the write at 0.4s gets overwritten
-      NonOverlappingUpdateResultJob.set(wait: (0.2 + i * 0.1).seconds).perform_later(@result, name: name, pause: 0.4.seconds)
+    # "000" isn't concurrency-limited, so it runs alongside A. Both read @result
+    # while it's still empty; "000" writes "s000c000" partway through A's run, but
+    # A pauses much longer and saves last of the two, overwriting it. A is
+    # scheduled well ahead of B–K so it reliably holds the semaphore first, and
+    # the rest of the chain builds only on A's clean write — so "000" never
+    # survives in the final result.
+    UpdateResultJob.set(wait: 0.1.seconds).perform_later(@result, name: "000", pause: 1.second)
+
+    NonOverlappingUpdateResultJob.set(wait: 0.1.seconds).perform_later(@result, name: "A", pause: 2.5.seconds)
+
+    ("B".."F").each_with_index do |name, i|
+      NonOverlappingUpdateResultJob.set(wait: (1 + i * 0.1).seconds).perform_later(@result, name: name, pause: 0.1.seconds)
     end
 
     ("G".."K").each_with_index do |name, i|
-      NonOverlappingUpdateResultJob.set(wait: (1 + i * 0.1).seconds).perform_later(@result, name: name)
+      NonOverlappingUpdateResultJob.set(wait: (1.5 + 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
   end
 
   test "run several jobs over the same record limiting concurrency" do
-    incr = 0
-    # C is the last one to update the record
-    # A: 0 to 0.5
-    # B: 0 to 1.0
-    # C: 0 to 1.5
+    # ThrottledUpdateResultJob has a concurrency limit of 3, so A, B and C run
+    # together — all reading @result while it's still empty — and D–H block. A
+    # and B finish quickly, freeing slots that drain D–H; C reads the empty
+    # status and pauses far longer than everyone else, so it saves last and its
+    # write (built on the empty status) overwrites all the others, leaving "C".
     assert_no_difference -> { SolidQueue::BlockedExecution.count } do
-      ("A".."C").each do |name|
-        ThrottledUpdateResultJob.perform_later(@result, name: name, pause: (0.5 + incr).seconds)
-        incr += 0.5
-      end
+      ThrottledUpdateResultJob.perform_later(@result, name: "A", pause: 0.5.seconds)
+      ThrottledUpdateResultJob.perform_later(@result, name: "B", pause: 0.5.seconds)
+      ThrottledUpdateResultJob.perform_later(@result, name: "C", pause: 3.seconds)
     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)
-    # These would finish all before B and C
+    wait_for(timeout: 2.seconds) { SolidQueue::ClaimedExecution.count >= 3 }
+
     assert_difference -> { SolidQueue::BlockedExecution.count }, +5 do
       ("D".."H").each do |name|
-        ThrottledUpdateResultJob.perform_later(@result, name: name, pause: 0.05.seconds)
+        ThrottledUpdateResultJob.perform_later(@result, name: name, pause: 0.01.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
-    # all other jobs, so it'll do the last update with only itself
     assert_stored_sequence(@result, [ "C" ])
   end
 
@@ -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
@@ -195,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")
@@ -255,9 +266,8 @@ class ConcurrencyControlsTest < ActiveSupport::TestCase
   private
     def assert_stored_sequence(result, sequence)
       expected = sequence.sort.map { |name| "s#{name}c#{name}" }.join
-      skip_active_record_query_cache do
-        assert_equal expected, result.reload.status.split(" + ").sort.join
-      end
+      actual = skip_active_record_query_cache { result.reload.status.split(" + ").sort.join }
+      assert_equal expected, actual
     end
 
     def wait_for_semaphores_to_be_released_for(timeout)

test/integration/forked_processes_lifecycle_test.rb

@@ -22,7 +22,7 @@ class ForkedProcessesLifecycleTest < ActiveSupport::TestCase
 
     wait_for_jobs_to_finish_for(2.seconds)
 
-    assert_equal 12, JobResult.count
+    assert_equal 12, skip_active_record_query_cache { JobResult.count }
     6.times { |i| assert_completed_job_results("job_#{i}", :background) }
     6.times { |i| assert_completed_job_results("job_#{i}", :default) }
 
@@ -124,7 +124,12 @@ class ForkedProcessesLifecycleTest < ActiveSupport::TestCase
     no_pause = enqueue_store_result_job("no pause")
     pause = enqueue_store_result_job("pause", pause: SolidQueue.shutdown_timeout + 10.seconds)
 
-    wait_while_with_timeout(1.second) { SolidQueue::ReadyExecution.count > 1 }
+    wait_while_with_timeout(5.seconds) {
+      SolidQueue::ReadyExecution.joins(:job).exists?(solid_queue_jobs: { active_job_id: pause.job_id })
+    }
+    wait_while_with_timeout(5.seconds) {
+      !JobResult.exists?(status: "started", value: "pause")
+    }
 
     signal_process(@pid, :TERM, wait: 0.5.second)
     wait_for_jobs_to_finish_for(2.seconds, except: pause)
@@ -167,19 +172,18 @@ class ForkedProcessesLifecycleTest < ActiveSupport::TestCase
   end
 
   test "process a job that exits" do
-    2.times do
-      enqueue_store_result_job("no exit", :background)
-      enqueue_store_result_job("no exit", :default)
-    end
+    # Enqueue all four "no exit" background jobs ahead of the exiting job. Jobs are
+    # claimed in enqueue order, so these are always claimed no later than exit_job
+    # and — being pauseless — finish before the worker exits, completing normally.
+    4.times { enqueue_store_result_job("no exit", :background) }
+    2.times { enqueue_store_result_job("no exit", :default) }
     enqueue_store_result_job("paused no exit", :default, pause: 0.5.second)
+
     # the worker for :background queue will exit abnormally
     exit_job = enqueue_store_result_job("exit", :background, exit_value: 1, pause: 0.5.second)
-    # this will run *after* exit_job (pause: 1.second) - should also be marked as failed
+    # claimed alongside exit_job and still paused when it exits, so it's failed too
     pause_job = enqueue_store_result_job("exit", :background, pause: 1.second)
 
-    # this will run *before* exit_job (no pause) - should complete normally
-    2.times { enqueue_store_result_job("no exit", :background) }
-
     wait_for_jobs_to_finish_for(3.seconds, except: [ exit_job, pause_job ])
 
     assert_completed_job_results("no exit", :default, 2)
@@ -236,6 +240,10 @@ class ForkedProcessesLifecycleTest < ActiveSupport::TestCase
     enqueue_store_result_job("pause", :default, pause: 0.5.seconds)
 
     wait_for_jobs_to_finish_for(1.second, except: [ killed_pause ])
+    # Ensure the long job has written its "started" row before we SIGKILL the worker.
+    wait_while_with_timeout(2.seconds) do
+      JobResult.where(status: "started", value: "killed_pause").none?
+    end
 
     worker = find_processes_registered_as("Worker").detect { |process| process.metadata["queues"].include? "background" }
     signal_process(worker.pid, :KILL, wait: 0.5.seconds)
@@ -298,15 +306,17 @@ def enqueue_store_result_job(value, queue_name = :background, **options)
     end
 
     def assert_completed_job_results(value, queue_name = :background, count = 1)
-      skip_active_record_query_cache do
-        assert_equal count, JobResult.where(queue_name: queue_name, status: "completed", value: value).count
-      end
+      actual = skip_active_record_query_cache {
+        JobResult.where(queue_name: queue_name, status: "completed", value: value).count
+      }
+      assert_equal count, actual
     end
 
     def assert_started_job_result(value, queue_name = :background, count = 1)
-      skip_active_record_query_cache do
-        assert_equal count, JobResult.where(queue_name: queue_name, status: "started", value: value).count
-      end
+      actual = skip_active_record_query_cache {
+        JobResult.where(queue_name: queue_name, status: "started", value: value).count
+      }
+      assert_equal count, actual
     end
 
     def assert_job_status(active_job, status)

test/integration/jobs_lifecycle_test.rb

@@ -40,11 +40,12 @@ 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 }
 
     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

@@ -14,22 +14,23 @@ class RecurringTasksTest < ActiveSupport::TestCase
   end
 
   test "enqueue and process periodic tasks" do
-    wait_for_jobs_to_be_enqueued(2, timeout: 2.5.seconds)
-    wait_for_jobs_to_finish_for(2.5.seconds)
+    wait_for_jobs_to_be_enqueued(2, timeout: 5.seconds)
+    wait_while_with_timeout(5.seconds) do
+      JobResult.where(status: "custom_status", value: "42").count < 2
+    end
 
     skip_active_record_query_cache do
       assert SolidQueue::Job.count >= 2
       SolidQueue::Job.all.each do |job|
         assert_equal "periodic_store_result", job.recurring_execution.task_key
         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
     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 = skip_active_record_query_cache { JobResult.where(status: "custom_status", value: "42").to_a }
+    assert_operator results.size, :>=, 2
   end
 
   test "persist and delete configured tasks" do

test/models/solid_queue/job_test.rb

@@ -256,7 +256,7 @@ class DiscardableNonOverlappingGroupedJob2 < NonOverlappingJob
     job = SolidQueue::Job.last
 
     worker = SolidQueue::Worker.new(queues: "background").tap(&:start)
-    sleep(0.2)
+    wait_while_with_timeout(2.seconds) { !job.reload.claimed? }
 
     assert_no_difference -> { SolidQueue::Job.count }, -> { SolidQueue::ClaimedExecution.count } do
       assert_raises SolidQueue::Execution::UndiscardableError do

test/test_helper.rb

@@ -102,7 +102,12 @@ def wait_while_with_timeout!(timeout, &block)
     # by the cached queries might have been updated, created or deleted in the forked
     # processes.
     def skip_active_record_query_cache(&block)
-      SolidQueue::Record.uncached(&block)
+      # Forked processes write to both the queue tables and the app's JobResult
+      # rows, which live in separate databases/connections, so bypass the query
+      # cache on both — SolidQueue::Record alone leaves JobResult reads stale.
+      SolidQueue::Record.uncached do
+        JobResult.uncached(&block)
+      end
     end
 
     # Silences specified exceptions during the execution of a block

test/unit/async_supervisor_test.rb

@@ -52,7 +52,9 @@ class AsyncSupervisorTest < ActiveSupport::TestCase
     wait_for_registered_processes(2, timeout: 3.seconds) # supervisor + 1 worker
     assert_registered_processes(kind: "Supervisor(async)")
 
-    wait_while_with_timeout(1.second) { SolidQueue::ClaimedExecution.count > 0 }
+    wait_while_with_timeout(5.seconds) {
+      SolidQueue::ClaimedExecution.count > 0 || SolidQueue::FailedExecution.count < 3
+    }
 
     skip_active_record_query_cache do
       assert_equal 0, SolidQueue::ClaimedExecution.count
@@ -74,7 +76,9 @@ class AsyncSupervisorTest < ActiveSupport::TestCase
     wait_for_registered_processes(2, timeout: 3.seconds) # supervisor + 1 worker
     assert_registered_processes(kind: "Supervisor(async)")
 
-    wait_while_with_timeout(1.second) { SolidQueue::ClaimedExecution.count > 0 }
+    wait_while_with_timeout(5.seconds) {
+      SolidQueue::ClaimedExecution.count > 0 || SolidQueue::FailedExecution.count < 3
+    }
 
     terminate_process(pid)