From 65b5a9297f3c279804ec285fb221153ded645712 Mon Sep 17 00:00:00 2001 From: Pissardo Date: Wed, 22 Jul 2026 11:12:17 +0200 Subject: [PATCH 1/2] Bypass JobResult's query cache in skip_active_record_query_cache MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Forked processes write to both the queue tables (SolidQueue::Record, on the queue database) and the app's JobResult rows (ApplicationRecord, on the primary database). skip_active_record_query_cache only bypassed the queue connection's query cache, so JobResult COUNT/find reads stayed cached and stale across forks — a source of intermittent integration-test failures. Nest JobResult.uncached so both connections' caches are bypassed. wait_for and wait_while_with_timeout use this helper, so they're covered too. --- test/test_helper.rb | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/test/test_helper.rb b/test/test_helper.rb index db5bd5c34..99d00ba27 100644 --- a/test/test_helper.rb +++ b/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 From 3faeaef31f132e158ece24ddf7fe84cdf38dc0a8 Mon Sep 17 00:00:00 2001 From: Pissardo Date: Wed, 22 Jul 2026 11:12:17 +0200 Subject: [PATCH 2/2] Reduce flaky integration test failures from timing and PK reuse Harden integration tests that fail intermittently on CI: - Wait for prior supervisors to deregister and clear leftover records before creating JobResult rows, avoiding recycled-PK overwrites from orphaned writers. - Replace fixed sleeps with wait_* conditions and lengthen timeouts. - Scope recurring-task assertions to the task's own results. - Retune the timing-sensitive tests so their intended results hold deterministically rather than relaxing the assertions: the "000" overwrite and the limiting-concurrency "C"-wins cases (widened pause margins), and "process a job that exits" (enqueue all "no exit" jobs before the exiting one so they always complete first). --- .../async_processes_lifecycle_test.rb | 18 +++-- test/integration/concurrency_controls_test.rb | 76 +++++++++++-------- .../forked_processes_lifecycle_test.rb | 42 ++++++---- test/integration/jobs_lifecycle_test.rb | 3 +- test/integration/recurring_tasks_test.rb | 17 +++-- test/models/solid_queue/job_test.rb | 2 +- test/unit/async_supervisor_test.rb | 8 +- 7 files changed, 97 insertions(+), 69 deletions(-) diff --git a/test/integration/async_processes_lifecycle_test.rb b/test/integration/async_processes_lifecycle_test.rb index fd284210e..6f269e220 100644 --- a/test/integration/async_processes_lifecycle_test.rb +++ b/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) diff --git a/test/integration/concurrency_controls_test.rb b/test/integration/concurrency_controls_test.rb index 4323865f2..01fdf8a11 100644 --- a/test/integration/concurrency_controls_test.rb +++ b/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) diff --git a/test/integration/forked_processes_lifecycle_test.rb b/test/integration/forked_processes_lifecycle_test.rb index 40495c20a..40ade6910 100644 --- a/test/integration/forked_processes_lifecycle_test.rb +++ b/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) diff --git a/test/integration/jobs_lifecycle_test.rb b/test/integration/jobs_lifecycle_test.rb index 8444c3759..4b23de296 100644 --- a/test/integration/jobs_lifecycle_test.rb +++ b/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 diff --git a/test/integration/recurring_tasks_test.rb b/test/integration/recurring_tasks_test.rb index f2fc7145f..c535ff058 100644 --- a/test/integration/recurring_tasks_test.rb +++ b/test/integration/recurring_tasks_test.rb @@ -14,8 +14,10 @@ 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 @@ -23,13 +25,12 @@ class RecurringTasksTest < ActiveSupport::TestCase 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 diff --git a/test/models/solid_queue/job_test.rb b/test/models/solid_queue/job_test.rb index 47702bd18..a776f703d 100644 --- a/test/models/solid_queue/job_test.rb +++ b/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 diff --git a/test/unit/async_supervisor_test.rb b/test/unit/async_supervisor_test.rb index d8843089a..962c4de7e 100644 --- a/test/unit/async_supervisor_test.rb +++ b/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)