Fetch pending builds with denormalized shared runners

app/models/ci/pending_build.rb

@@ -10,6 +10,7 @@ module Ci
 
     scope :ref_protected, -> { where(protected: true) }
     scope :queued_before, ->(time) { where(arel_table[:created_at].lt(time)) }
+    scope :with_instance_runners, -> { where(instance_runners_enabled: true) }
 
     def self.upsert_from_build!(build)
       entry = self.new(args_from_build(build))

app/services/ci/queue/builds_table_strategy.rb

@@ -53,6 +53,10 @@ module Ci
         relation.pluck(:id)
       end
 
+      def use_denormalized_shared_runners_data?
+        false
+      end
+
       private
 
       def running_builds_for_shared_runners

app/services/ci/queue/pending_builds_strategy.rb

@@ -11,25 +11,9 @@ module Ci
 
       # rubocop:disable CodeReuse/ActiveRecord
       def builds_for_shared_runner
-        relation = new_builds
-          # don't run projects which have not enabled shared runners and builds
-          .joins('INNER JOIN projects ON ci_pending_builds.project_id = projects.id')
-          .where(projects: { shared_runners_enabled: true, pending_delete: false })
-          .joins('LEFT JOIN project_features ON ci_pending_builds.project_id = project_features.project_id')
-          .where('project_features.builds_access_level IS NULL or project_features.builds_access_level > 0')
+        shared_builds = builds_available_for_shared_runners
 
-        if Feature.enabled?(:ci_queueing_disaster_recovery_disable_fair_scheduling, runner, type: :ops, default_enabled: :yaml)
-          # if disaster recovery is enabled, we fallback to FIFO scheduling
-          relation.order('ci_pending_builds.build_id ASC')
-        else
-          # Implement fair scheduling
-          # this returns builds that are ordered by number of running builds
-          # we prefer projects that don't use shared runners at all
-          relation
-            .with(running_builds_for_shared_runners_cte.to_arel)
-            .joins("LEFT JOIN project_builds ON ci_pending_builds.project_id = project_builds.project_id")
-            .order(Arel.sql('COALESCE(project_builds.running_builds, 0) ASC'), 'ci_pending_builds.build_id ASC')
-        end
+        builds_ordered_for_shared_runners(shared_builds)
       end
 
       def builds_matching_tag_ids(relation, ids)
@@ -52,8 +36,40 @@ module Ci
         relation.pluck(:build_id)
       end
 
+      def use_denormalized_shared_runners_data?
+        ::Feature.enabled?(:ci_queueing_denormalize_shared_runners_information, runner, type: :development, default_enabled: :yaml)
+      end
+
       private
 
+      def builds_available_for_shared_runners
+        if use_denormalized_shared_runners_data?
+          new_builds.with_instance_runners
+        else
+          new_builds
+            # don't run projects which have not enabled shared runners and builds
+            .joins('INNER JOIN projects ON ci_pending_builds.project_id = projects.id')
+            .where(projects: { shared_runners_enabled: true, pending_delete: false })
+            .joins('LEFT JOIN project_features ON ci_pending_builds.project_id = project_features.project_id')
+            .where('project_features.builds_access_level IS NULL or project_features.builds_access_level > 0')
+        end
+      end
+
+      def builds_ordered_for_shared_runners(relation)
+        if Feature.enabled?(:ci_queueing_disaster_recovery_disable_fair_scheduling, runner, type: :ops, default_enabled: :yaml)
+          # if disaster recovery is enabled, we fallback to FIFO scheduling
+          relation.order('ci_pending_builds.build_id ASC')
+        else
+          # Implement fair scheduling
+          # this returns builds that are ordered by number of running builds
+          # we prefer projects that don't use shared runners at all
+          relation
+            .with(running_builds_for_shared_runners_cte.to_arel)
+            .joins("LEFT JOIN project_builds ON ci_pending_builds.project_id = project_builds.project_id")
+            .order(Arel.sql('COALESCE(project_builds.running_builds, 0) ASC'), 'ci_pending_builds.build_id ASC')
+        end
+      end
+
       def running_builds_for_shared_runners_cte
         running_builds = ::Ci::RunningBuild
           .instance_type

config/feature_flags/development/ci_queueing_denormalize_shared_runners_information.yml

+---
+name: ci_queueing_denormalize_shared_runners_information
+introduced_by_url:
+rollout_issue_url:
+milestone: '14.2'
+type: development
+group: group::pipeline execution
+default_enabled: false

ee/app/services/ee/ci/queue/build_queue_service.rb

@@ -19,6 +19,13 @@ module EE
 
         # rubocop: disable CodeReuse/ActiveRecord
         def enforce_minutes_based_on_cost_factors(relation)
+          if strategy.use_denormalized_shared_runners_data?
+            # If shared runners information is denormalized then the query does not include the join
+            # with `projects` anymore, so we need to add it until we use denormalized ci minutes
+
+            relation = relation.joins('INNER JOIN projects ON ci_pending_builds.project_id = projects.id')
+          end
+
           visibility_relation = ::CommitStatus.where(
             projects: { visibility_level: runner.visibility_levels_without_minutes_quota })
 

ee/spec/services/ci/register_job_service_spec.rb

@@ -281,6 +281,20 @@ RSpec.describe Ci::RegisterJobService, '#execute' do
       stub_feature_flags(ci_pending_builds_queue_source: true)
     end
 
+    context 'with ci_queueing_denormalize_shared_runners_information enabled' do
+      before do
+        stub_feature_flags(ci_queueing_denormalize_shared_runners_information: true)
+      end
+
+      include_examples 'namespace minutes quota'
+    end
+
+    context 'with ci_queueing_denormalize_shared_runners_information disabled' do
+      before do
+        stub_feature_flags(ci_queueing_denormalize_shared_runners_information: false)
+      end
+
       include_examples 'namespace minutes quota'
     end
+  end
 end

spec/models/ci/pending_build_spec.rb

@@ -14,6 +14,28 @@ RSpec.describe Ci::PendingBuild do
     it { is_expected.to belong_to :namespace }
   end
 
+  describe 'scopes' do
+    describe '.with_instance_runners' do
+      subject(:pending_builds) { described_class.with_instance_runners }
+
+      let!(:pending_build_1) { create(:ci_pending_build, instance_runners_enabled: false) }
+
+      context 'when pending builds cannot be picked up by runner' do
+        it 'returns an empty collection of pending builds' do
+          expect(pending_builds).to be_empty
+        end
+      end
+
+      context 'when pending builds can be picked up by runner' do
+        let!(:pending_build_2) { create(:ci_pending_build) }
+
+        it 'returns matching pending builds' do
+          expect(pending_builds).to contain_exactly(pending_build_2)
+        end
+      end
+    end
+  end
+
   describe '.upsert_from_build!' do
     context 'another pending entry does not exist' do
       it 'creates a new pending entry' do

spec/services/ci/register_job_service_spec.rb

@@ -90,6 +90,9 @@ module Ci
         context 'allow shared runners' do
           before do
             project.update!(shared_runners_enabled: true)
+            pipeline.reload
+            pending_job.reload
+            pending_job.create_queuing_entry!
           end
 
           context 'for multiple builds' do
@@ -703,9 +706,23 @@ module Ci
           stub_feature_flags(ci_pending_builds_queue_source: true)
         end
 
+        context 'with ci_queueing_denormalize_shared_runners_information enabled' do
+          before do
+            stub_feature_flags(ci_queueing_denormalize_shared_runners_information: true)
+          end
+
           include_examples 'handles runner assignment'
         end
 
+        context 'with ci_queueing_denormalize_shared_runners_information disabled' do
+          before do
+            stub_feature_flags(ci_queueing_denormalize_shared_runners_information: false)
+          end
+
+          include_examples 'handles runner assignment'
+        end
+      end
+
       context 'when not using pending builds table' do
         before do
           stub_feature_flags(ci_pending_builds_queue_source: false)
@@ -777,6 +794,11 @@ module Ci
       end
 
       context 'when shared runner is used' do
+        before do
+          pending_job.reload
+          pending_job.create_queuing_entry!
+        end
+
         let(:runner) { create(:ci_runner, :instance, tag_list: %w(tag1 tag2)) }
         let(:expected_shared_runner) { true }
         let(:expected_shard) { ::Gitlab::Ci::Queue::Metrics::DEFAULT_METRICS_SHARD }