Merge branch 'main' of github.com:rails/solid_queue into deprecated-retries-option

Author: p-schlickmann

test/dummy/app/jobs/infinite_recursion_job.rb

@@ -3,6 +3,8 @@ class InfiniteRecursionJob < ApplicationJob
 
   def perform
     start
+  rescue SystemStackError => e
+    raise ExpectedTestError, "stack level too deep", e.backtrace
   end
 
   private

test/dummy/app/jobs/sequential_update_result_job.rb renamed to test/dummy/app/jobs/non_overlapping_update_result_job.rb

@@ -1,3 +1,3 @@
-class SequentialUpdateResultJob < UpdateResultJob
+class NonOverlappingUpdateResultJob < UpdateResultJob
   limits_concurrency key: ->(job_result, **) { job_result }
 end

test/dummy/app/jobs/update_result_job.rb

@@ -1,5 +1,6 @@
 class UpdateResultJob < ApplicationJob
   def perform(job_result, name:, pause: nil, exception: nil)
+    job_result.status += " + " unless job_result.status.blank?
     job_result.status += "s#{name}"
 
     sleep(pause) if pause

test/integration/concurrency_controls_test.rb

@@ -6,7 +6,7 @@ class ConcurrencyControlsTest < ActiveSupport::TestCase
   self.use_transactional_tests = false
 
   setup do
-    @result = JobResult.create!(queue_name: "default", status: "seq: ")
+    @result = JobResult.create!(queue_name: "default", status: "")
 
     default_worker = { queues: "default", polling_interval: 0.1, processes: 3, threads: 2 }
     dispatcher = { polling_interval: 0.1, batch_size: 200, concurrency_maintenance_interval: 1 }
@@ -20,13 +20,13 @@ class ConcurrencyControlsTest < ActiveSupport::TestCase
     terminate_process(@pid) if process_exists?(@pid)
   end
 
-  test "run several conflicting jobs over the same record sequentially" do
+  test "run several conflicting jobs over the same record without overlapping" do
     ("A".."F").each do |name|
-      SequentialUpdateResultJob.perform_later(@result, name: name, pause: 0.2.seconds)
+      NonOverlappingUpdateResultJob.perform_later(@result, name: name, pause: 0.2.seconds)
     end
 
     ("G".."K").each do |name|
-      SequentialUpdateResultJob.perform_later(@result, name: name)
+      NonOverlappingUpdateResultJob.perform_later(@result, name: name)
     end
 
     wait_for_jobs_to_finish_for(5.seconds)
@@ -39,11 +39,11 @@ class ConcurrencyControlsTest < ActiveSupport::TestCase
     UpdateResultJob.set(wait: 0.23.seconds).perform_later(@result, name: "000", pause: 0.1.seconds)
 
     ("A".."F").each_with_index do |name, i|
-      SequentialUpdateResultJob.set(wait: (0.2 + i * 0.01).seconds).perform_later(@result, name: name, pause: 0.3.seconds)
+      NonOverlappingUpdateResultJob.set(wait: (0.2 + i * 0.01).seconds).perform_later(@result, name: name, pause: 0.3.seconds)
     end
 
     ("G".."K").each_with_index do |name, i|
-      SequentialUpdateResultJob.set(wait: (0.3 + i * 0.01).seconds).perform_later(@result, name: name)
+      NonOverlappingUpdateResultJob.set(wait: (0.3 + i * 0.01).seconds).perform_later(@result, name: name)
     end
 
     wait_for_jobs_to_finish_for(5.seconds)
@@ -85,11 +85,11 @@ class ConcurrencyControlsTest < ActiveSupport::TestCase
   test "run several jobs over the same record sequentially, with some of them failing" do
     ("A".."F").each_with_index do |name, i|
       # A, C, E will fail, for i= 0, 2, 4
-      SequentialUpdateResultJob.perform_later(@result, name: name, pause: 0.2.seconds, exception: (ExpectedTestError if i.even?))
+      NonOverlappingUpdateResultJob.perform_later(@result, name: name, pause: 0.2.seconds, exception: (ExpectedTestError if i.even?))
     end
 
     ("G".."K").each do |name|
-      SequentialUpdateResultJob.perform_later(@result, name: name)
+      NonOverlappingUpdateResultJob.perform_later(@result, name: name)
     end
 
     wait_for_jobs_to_finish_for(5.seconds)
@@ -100,7 +100,7 @@ class ConcurrencyControlsTest < ActiveSupport::TestCase
 
   test "rely on dispatcher to unblock blocked executions with an available semaphore" do
     # Simulate a scenario where we got an available semaphore and some stuck jobs
-    job = SequentialUpdateResultJob.perform_later(@result, name: "A")
+    job = NonOverlappingUpdateResultJob.perform_later(@result, name: "A")
 
     wait_for_jobs_to_finish_for(5.seconds)
     assert_no_unfinished_jobs
@@ -114,7 +114,7 @@ class ConcurrencyControlsTest < ActiveSupport::TestCase
     # Now enqueue more jobs under that same key. They'll be all locked
     assert_difference -> { SolidQueue::BlockedExecution.count }, +10 do
       ("B".."K").each do |name|
-        SequentialUpdateResultJob.perform_later(@result, name: name)
+        NonOverlappingUpdateResultJob.perform_later(@result, name: name)
       end
     end
 
@@ -127,14 +127,12 @@ class ConcurrencyControlsTest < ActiveSupport::TestCase
     wait_for_jobs_to_finish_for(5.seconds)
     assert_no_unfinished_jobs
 
-    # We can't ensure the order between B and C, because it depends on which worker wins when
-    # unblocking, as one will try to unblock B and another C
-    assert_stored_sequence @result, ("A".."K").to_a, [ "A", "C", "B" ] + ("D".."K").to_a
+    assert_stored_sequence @result, ("A".."K").to_a
   end
 
   test "rely on dispatcher to unblock blocked executions with an expired semaphore" do
     # Simulate a scenario where we got an available semaphore and some stuck jobs
-    job = SequentialUpdateResultJob.perform_later(@result, name: "A")
+    job = NonOverlappingUpdateResultJob.perform_later(@result, name: "A")
     wait_for_jobs_to_finish_for(5.seconds)
     assert_no_unfinished_jobs
 
@@ -147,7 +145,7 @@ class ConcurrencyControlsTest < ActiveSupport::TestCase
     # Now enqueue more jobs under that same key. They'll be all locked
     assert_difference -> { SolidQueue::BlockedExecution.count }, +10 do
       ("B".."K").each do |name|
-        SequentialUpdateResultJob.perform_later(@result, name: name)
+        NonOverlappingUpdateResultJob.perform_later(@result, name: name)
       end
     end
 
@@ -159,13 +157,11 @@ class ConcurrencyControlsTest < ActiveSupport::TestCase
     wait_for_jobs_to_finish_for(5.seconds)
     assert_no_unfinished_jobs
 
-    # We can't ensure the order between B and C, because it depends on which worker wins when
-    # unblocking, as one will try to unblock B and another C
-    assert_stored_sequence @result, ("A".."K").to_a, [ "A", "C", "B" ] + ("D".."K").to_a
+    assert_stored_sequence @result, ("A".."K").to_a
   end
 
   test "don't block claimed executions that get released" do
-    SequentialUpdateResultJob.perform_later(@result, name: "I'll be released to ready", pause: SolidQueue.shutdown_timeout + 10.seconds)
+    NonOverlappingUpdateResultJob.perform_later(@result, name: "I'll be released to ready", pause: SolidQueue.shutdown_timeout + 10.seconds)
     job = SolidQueue::Job.last
 
     sleep(0.2)
@@ -184,8 +180,8 @@ class ConcurrencyControlsTest < ActiveSupport::TestCase
     skip if Rails::VERSION::MAJOR == 7 && Rails::VERSION::MINOR == 2
 
     ActiveRecord::Base.transaction do
-      SequentialUpdateResultJob.perform_later(@result, name: "A", pause: 0.2.seconds)
-      SequentialUpdateResultJob.perform_later(@result, name: "B")
+      NonOverlappingUpdateResultJob.perform_later(@result, name: "A", pause: 0.2.seconds)
+      NonOverlappingUpdateResultJob.perform_later(@result, name: "B")
 
       begin
         assert_equal 2, SolidQueue::Job.count
@@ -219,7 +215,7 @@ class ConcurrencyControlsTest < ActiveSupport::TestCase
   end
 
   test "discard on conflict across different concurrency keys" do
-    another_result = JobResult.create!(queue_name: "default", status: "seq: ")
+    another_result = JobResult.create!(queue_name: "default", status: "")
     DiscardableUpdateResultJob.perform_later(@result, name: "1", pause: 0.2)
     DiscardableUpdateResultJob.perform_later(another_result, name: "2", pause: 0.2)
     DiscardableUpdateResultJob.perform_later(@result, name: "3") # Should be discarded
@@ -239,6 +235,8 @@ class ConcurrencyControlsTest < ActiveSupport::TestCase
     DiscardableUpdateResultJob.perform_later(@result, name: "2")
 
     wait_for_jobs_to_finish_for(5.seconds)
+    wait_for_semaphores_to_be_released_for(2.seconds)
+
     assert_no_unfinished_jobs
 
     # Enqueue another job that shouldn't be discarded or blocked
@@ -250,10 +248,18 @@ class ConcurrencyControlsTest < ActiveSupport::TestCase
   end
 
   private
-    def assert_stored_sequence(result, *sequences)
-      expected = sequences.map { |sequence| "seq: " + sequence.map { |name| "s#{name}c#{name}" }.join }
+    def assert_stored_sequence(result, sequence)
+      expected = sequence.sort.map { |name| "s#{name}c#{name}" }.join
       skip_active_record_query_cache do
-        assert_includes expected, result.reload.status
+        assert_equal expected, result.reload.status.split(" + ").sort.join
+      end
+    end
+
+    def wait_for_semaphores_to_be_released_for(timeout)
+      wait_while_with_timeout(timeout) do
+        skip_active_record_query_cache do
+          SolidQueue::Semaphore.available.invert_where.any?
+        end
       end
     end
 end

test/integration/instrumentation_test.rb

@@ -234,7 +234,7 @@ class InstrumentationTest < ActiveSupport::TestCase
     5.times { AddToBufferJob.perform_later("A") }
     # 1 ready + 3 blocked
     result = JobResult.create!
-    4.times { SequentialUpdateResultJob.perform_later(result, name: "A") }
+    4.times { NonOverlappingUpdateResultJob.perform_later(result, name: "A") }
 
     events = subscribed("discard_all.solid_queue") do
       SolidQueue::ReadyExecution.discard_all_from_jobs(SolidQueue::Job.all)
@@ -261,7 +261,7 @@ class InstrumentationTest < ActiveSupport::TestCase
   test "unblocking job emits release_blocked event" do
     result = JobResult.create!
     # 1 ready, 2 blocked
-    3.times { SequentialUpdateResultJob.perform_later(result, name: "A") }
+    3.times { NonOverlappingUpdateResultJob.perform_later(result, name: "A") }
 
     # Simulate expiry of the concurrency locks
     travel_to 3.days.from_now
@@ -283,11 +283,11 @@ class InstrumentationTest < ActiveSupport::TestCase
   test "unblocking jobs in bulk emits release_many_blocked event" do
     result = JobResult.create!
     # 1 ready, 3 blocked
-    4.times { SequentialUpdateResultJob.perform_later(result, name: "A") }
+    4.times { NonOverlappingUpdateResultJob.perform_later(result, name: "A") }
 
     # 1 ready, 2 blocked
     result = JobResult.create!
-    3.times { SequentialUpdateResultJob.perform_later(result, name: "B") }
+    3.times { NonOverlappingUpdateResultJob.perform_later(result, name: "B") }
 
     # Simulate expiry of the concurrency locks
     travel_to 3.days.from_now
@@ -309,7 +309,7 @@ class InstrumentationTest < ActiveSupport::TestCase
 
     events = subscribed("enqueue_recurring_task.solid_queue") do
       scheduler.start
-      sleep 1.01
+      wait_while_with_timeout(1.1.second) { SolidQueue::RecurringExecution.none? }
       scheduler.stop
     end
 

test/models/solid_queue/failed_execution_test.rb

@@ -15,13 +15,12 @@ class SolidQueue::FailedExecutionTest < ActiveSupport::TestCase
   end
 
   test "run job that fails with a SystemStackError (stack level too deep)" do
-    silence_on_thread_error_for(SystemStackError) do
-      InfiniteRecursionJob.perform_later
-      @worker.start
+    InfiniteRecursionJob.perform_later
+    @worker.start
 
-      assert_equal 1, SolidQueue::FailedExecution.count
-      assert SolidQueue::Job.last.failed?
-    end
+    assert_equal 1, SolidQueue::FailedExecution.count
+    assert SolidQueue::Job.last.failed?
+    assert_equal "stack level too deep", SolidQueue::FailedExecution.last.message
   end
 
   test "retry failed job" do

test/unit/process_recovery_test.rb

@@ -26,9 +26,11 @@ class ProcessRecoveryTest < ActiveSupport::TestCase
     worker_process = SolidQueue::Process.find_by(kind: "Worker")
     assert worker_process
 
-    # Enqueue a job and manually claim it for the worker to avoid timing races
-    job = enqueue_store_result_job(42)
-    claimed_execution = SolidQueue::ReadyExecution.claim("*", 5, worker_process.id).first
+    # Enqueue a job and wait for it to be claimed
+    StoreResultJob.perform_later(42, pause: 10.seconds)
+    wait_while_with_timeout(3.seconds) { SolidQueue::ClaimedExecution.none? }
+
+    claimed_execution = SolidQueue::ClaimedExecution.last
     assert claimed_execution.present?
     assert_equal worker_process.id, claimed_execution.process_id
 
@@ -40,9 +42,8 @@ class ProcessRecoveryTest < ActiveSupport::TestCase
     worker_pid = worker_process.pid
     terminate_process(worker_pid, signal: :KILL)
 
-
     # Wait for the supervisor to reap the worker and fail the job
-    wait_for_failed_executions(1, timeout: 5.seconds)
+    wait_while_with_timeout(3.seconds) { SolidQueue::FailedExecution.none? }
 
     # Assert the execution is failed
     failed_execution = SolidQueue::FailedExecution.last
@@ -53,32 +54,4 @@ class ProcessRecoveryTest < ActiveSupport::TestCase
     wait_for_registered_processes(2, timeout: 5.seconds)
     assert_operator SolidQueue::Process.where(kind: "Worker").count, :>=, 1
   end
-
-  private
-    def assert_registered_workers_for(*queues, supervisor_pid: nil)
-      workers = find_processes_registered_as("Worker")
-      registered_queues = workers.map { |process| process.metadata["queues"] }.compact
-      assert_equal queues.map(&:to_s).sort, registered_queues.sort
-      if supervisor_pid
-        assert_equal [ supervisor_pid ], workers.map { |process| process.supervisor.pid }.uniq
-      end
-    end
-
-    def enqueue_store_result_job(value, queue_name = :default, **options)
-      StoreResultJob.set(queue: queue_name).perform_later(value, **options)
-    end
-
-    def assert_no_claimed_jobs
-      skip_active_record_query_cache do
-        assert_empty SolidQueue::ClaimedExecution.all
-      end
-    end
-
-    def wait_for_claimed_executions(count, timeout: 1.second)
-      wait_for(timeout: timeout) { SolidQueue::ClaimedExecution.count == count }
-    end
-
-    def wait_for_failed_executions(count, timeout: 1.second)
-      wait_for(timeout: timeout) { SolidQueue::FailedExecution.count == count }
-    end
 end