diff --git a/test/integration/async_processes_lifecycle_test.rb b/test/integration/async_processes_lifecycle_test.rb index fd284210..e4f74ec6 100644 --- a/test/integration/async_processes_lifecycle_test.rb +++ b/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 diff --git a/test/integration/concurrency_controls_test.rb b/test/integration/concurrency_controls_test.rb index a12c48e4..9f748ea7 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 @@ -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,8 +76,10 @@ 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| @@ -76,7 +87,7 @@ class ConcurrencyControlsTest < ActiveSupport::TestCase 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") diff --git a/test/integration/jobs_lifecycle_test.rb b/test/integration/jobs_lifecycle_test.rb index 8444c375..53e8c97c 100644 --- a/test/integration/jobs_lifecycle_test.rb +++ b/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 diff --git a/test/integration/recurring_tasks_test.rb b/test/integration/recurring_tasks_test.rb index f2fc7145..64059e5a 100644 --- a/test/integration/recurring_tasks_test.rb +++ b/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