Merge branch 'disable-jobwaiter-for-many-jobs' into 'master'

Disable JobWaiter when there are many jobs

See merge request gitlab-org/gitlab!27565

app/workers/concerns/waitable_worker.rb

@@ -9,6 +9,15 @@ module WaitableWorker
       # Short-circuit: it's more efficient to do small numbers of jobs inline
       return bulk_perform_inline(args_list) if args_list.size <= 3
 
+      # Don't wait if there's too many jobs to be waited for. Not including the
+      # waiter allows them to be deduplicated and it skips waiting for jobs that
+      # are not likely to finish within the timeout. This assumes we can process
+      # 10 jobs per second:
+      # https://gitlab.com/gitlab-com/gl-infra/scalability/-/issues/205
+      if ::Feature.enabled?(:skip_job_waiter_for_large_batches)
+        return bulk_perform_async(args_list) if args_list.length >= 10 * timeout
+      end
+
       waiter = Gitlab::JobWaiter.new(args_list.size, worker_label: self.to_s)
 
       # Point all the bulk jobs at the same JobWaiter. Converts, [[1], [2], [3]]

spec/workers/concerns/waitable_worker_spec.rb

@@ -44,11 +44,35 @@ describe WaitableWorker do
       expect(worker.counter).to eq(6)
     end
 
-    it 'runs > 3 jobs using sidekiq' do
+    it 'runs > 3 jobs using sidekiq and a waiter key' do
       expect(worker).to receive(:bulk_perform_async)
+                          .with([[1, anything], [2, anything], [3, anything], [4, anything]])
 
       worker.bulk_perform_and_wait([[1], [2], [3], [4]])
     end
+
+    it 'runs > 10 * timeout jobs using sidekiq and no waiter key' do
+      arguments = 1.upto(21).map { |i| [i] }
+
+      expect(worker).to receive(:bulk_perform_async).with(arguments)
+
+      worker.bulk_perform_and_wait(arguments, timeout: 2)
+    end
+
+    context 'when the skip_job_waiter_for_large_batches flag is disabled' do
+      before do
+        stub_feature_flags(skip_job_waiter_for_large_batches: false)
+      end
+
+      it 'runs jobs over 10 * the timeout using a waiter key' do
+        arguments = 1.upto(21).map { |i| [i] }
+        arguments_with_waiter = arguments.map { |arg| arg + [anything] }
+
+        expect(worker).to receive(:bulk_perform_async).with(arguments_with_waiter)
+
+        worker.bulk_perform_and_wait(arguments, timeout: 2)
+      end
+    end
   end
 
   describe '.bulk_perform_inline' do