Attempt reduce flakiness of processes_lifecycle_test.rb

Author: p-schlickmann

test/integration/processes_lifecycle_test.rb

@@ -123,22 +123,22 @@ class ProcessesLifecycleTest < ActiveSupport::TestCase
     no_pause = enqueue_store_result_job("no pause")
     pause = enqueue_store_result_job("pause", pause: SolidQueue.shutdown_timeout + 10.second)
 
-    wait_while_with_timeout(1.second) { SolidQueue::ReadyExecution.count > 0 }
+    wait_while_with_timeout(1.second) { SolidQueue::ReadyExecution.count > 1 }
 
-    signal_process(@pid, :TERM, wait: 0.5)
+    signal_process(@pid, :TERM, wait: 0.5.seconds)
 
-    sleep(SolidQueue.shutdown_timeout + 0.5.second)
+    t = SolidQueue.shutdown_timeout + 1.second
 
-    assert_completed_job_results("no pause")
-    assert_job_status(no_pause, :finished)
+    assert_job_status(no_pause, :finished, timeout: t)
+    assert_completed_job_results("no pause", timeout: t)
+
+    # This job was left claimed as the worker was shutdown without a chance to terminate orderly
+    assert_job_status(pause, :claimed, timeout: t)
+    assert_started_job_result("pause", timeout: t)
 
-    # This job was left claimed as the worker was shutdown without
-    # a chance to terminate orderly
-    assert_started_job_result("pause")
-    assert_job_status(pause, :claimed)
 
     # The process running the long job couldn't deregister, the other did
-    assert_registered_workers_for(:background)
+    wait_for_registered_processes(1, timeout: 3.seconds)
 
     # Now wait until the supervisor finishes for real, which will complete the cleanup
     wait_for_process_termination_with_timeout(@pid, timeout: 1.second)
@@ -300,27 +300,50 @@ def enqueue_store_result_job(value, queue_name = :background, **options)
       StoreResultJob.set(queue: queue_name).perform_later(value, **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
+    def assert_completed_job_results(value, queue_name = :background, count = 1, timeout: nil)
+      if timeout
+        wait_while_with_timeout!(timeout) { not check_job_result(value, "completed", queue_name, count) }
+      else
+        check_job_result(value, "completed", queue_name, count)
+      end
+    end
+
+    def assert_started_job_result(value, queue_name = :background, count = 1, timeout: nil)
+      if timeout
+        wait_while_with_timeout!(timeout) { not check_job_result(value, "started", queue_name, count) }
+      else
+        check_job_result(value, "started", queue_name, count)
       end
     end
 
-    def assert_started_job_result(value, queue_name = :background, count = 1)
+    def check_job_result(value, status, 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
+        count == JobResult.where(queue_name: queue_name, status: status, value: value).count
+      end
+    end
+
+    def assert_job_status(active_job, status, timeout: nil)
+      # Make sure we skip AR query cache. Otherwise the queries done here
+      # might be cached and since we haven't done any non-SELECT queries
+      # after they were cached on the connection used in the test, the cache
+      # will still apply, even though the data returned by the cached queries
+      # might have been deleted in the forked processes.
+      if timeout
+        wait_while_with_timeout!(timeout) { not check_job_status(active_job, status)}
+      else
+        assert check_job_status(active_job, status)
       end
     end
 
-    def assert_job_status(active_job, status)
+    def check_job_status(active_job, status)
       # Make sure we skip AR query cache. Otherwise the queries done here
       # might be cached and since we haven't done any non-SELECT queries
       # after they were cached on the connection used in the test, the cache
       # will still apply, even though the data returned by the cached queries
       # might have been deleted in the forked processes.
       skip_active_record_query_cache do
         job = SolidQueue::Job.find_by(active_job_id: active_job.job_id)
-        assert job.public_send("#{status}?")
+        job.public_send("#{status}?")
       end
     end
 end