Use pending build table as a source for builds queue

app/models/ci/build.rb

@@ -178,25 +178,6 @@ module Ci
       joins(:metadata).where("ci_builds_metadata.config_options -> 'artifacts' -> 'reports' ?| array[:job_types]", job_types: job_types)
     end
 
-    scope :matches_tag_ids, -> (tag_ids) do
-      matcher = ::ActsAsTaggableOn::Tagging
-        .where(taggable_type: CommitStatus.name)
-        .where(context: 'tags')
-        .where('taggable_id = ci_builds.id')
-        .where.not(tag_id: tag_ids).select('1')
-
-      where("NOT EXISTS (?)", matcher)
-    end
-
-    scope :with_any_tags, -> do
-      matcher = ::ActsAsTaggableOn::Tagging
-        .where(taggable_type: CommitStatus.name)
-        .where(context: 'tags')
-        .where('taggable_id = ci_builds.id').select('1')
-
-      where("EXISTS (?)", matcher)
-    end
-
     scope :queued_before, ->(time) { where(arel_table[:queued_at].lt(time)) }
 
     scope :preload_project_and_pipeline_project, -> do

app/models/ci/pending_build.rb

@@ -7,6 +7,8 @@ module Ci
     belongs_to :project
     belongs_to :build, class_name: 'Ci::Build'
 
+    scope :ref_protected, -> { where(protected: true) }
+
     def self.upsert_from_build!(build)
       entry = self.new(build: build, project: build.project, protected: build.protected?)
 

app/models/commit_status.rb

@@ -85,6 +85,25 @@ class CommitStatus < ApplicationRecord
     merge(or_conditions)
   end
 
+  scope :matches_tag_ids, ->(tag_ids) do
+    matcher = ::ActsAsTaggableOn::Tagging
+      .where(taggable_type: CommitStatus.name)
+      .where(context: 'tags')
+      .where('taggable_id = ci_builds.id')
+      .where.not(tag_id: tag_ids).select('1')
+
+    where("NOT EXISTS (?)", matcher)
+  end
+
+  scope :with_any_tags, -> do
+    matcher = ::ActsAsTaggableOn::Tagging
+      .where(taggable_type: CommitStatus.name)
+      .where(context: 'tags')
+      .where('taggable_id = ci_builds.id').select('1')
+
+    where("EXISTS (?)", matcher)
+  end
+
   # We use `Enums::Ci::CommitStatus.failure_reasons` here so that EE can more easily
   # extend this `Hash` with new values.
   enum_with_nil failure_reason: Enums::Ci::CommitStatus.failure_reasons

app/services/ci/register_job_service.rb

@@ -113,11 +113,11 @@ module Ci
         end
 
       # pick builds that does not have other tags than runner's one
-      builds = builds.matches_tag_ids(runner.tags.ids)
+      builds = builds.merge(::CommitStatus.matches_tag_ids(runner.tags.ids))
 
       # pick builds that have at least one tag
       unless runner.run_untagged?
-        builds = builds.with_any_tags
+        builds = builds.merge(::CommitStatus.with_any_tags)
       end
 
       # pick builds that older than specified age
@@ -125,7 +125,7 @@ module Ci
         builds = builds.queued_before(params[:job_age].seconds.ago)
       end
 
-      build_ids = retrieve_queue(-> { builds.pluck(:id) })
+      build_ids = retrieve_queue(-> { builds.map(&:id) })
 
       @metrics.observe_queue_size(-> { build_ids.size }, @runner.runner_type)
 
@@ -261,9 +261,10 @@ module Ci
 
     # rubocop: disable CodeReuse/ActiveRecord
     def builds_for_shared_runner
-      relation = new_builds.
+      relation = new_builds
         # don't run projects which have not enabled shared runners and builds
-        joins(:project).where(projects: { shared_runners_enabled: true, pending_delete: false })
+        .joins('INNER JOIN projects ON ci_builds.project_id = projects.id')
+        .where(projects: { shared_runners_enabled: true, pending_delete: false })
         .joins('LEFT JOIN project_features ON ci_builds.project_id = project_features.project_id')
         .where('project_features.builds_access_level IS NULL or project_features.builds_access_level > 0')
 
@@ -280,8 +281,12 @@ module Ci
       end
     end
 
+    def new_builds_in_projects(scope)
+      new_builds.where("project_id IN (#{scope.select(:id).to_sql})").order('id ASC')
+    end
+
     def builds_for_project_runner
-      new_builds.where(project: runner.projects.without_deleted.with_builds_enabled).order('id ASC')
+      new_builds_in_projects(runner.projects.without_deleted.with_builds_enabled)
     end
 
     def builds_for_group_runner
@@ -293,27 +298,41 @@ module Ci
         .with_group_runners_enabled
         .with_builds_enabled
         .without_deleted
-      new_builds.where(project: projects).order('id ASC')
+
+      new_builds_in_projects(projects)
     end
 
-    def running_builds_for_shared_runners
-      Ci::Build.running.where(runner: Ci::Runner.instance_type)
-        .group(:project_id).select(:project_id, 'count(*) AS running_builds')
+    def all_running_builds
+      if Feature.enabled?(:ci_pending_builds_queue_source, runner, default_enabled: :yaml)
+        Ci::RunningBuild.all
+      else
+        Ci::Build.running
+      end
     end
 
     def all_builds
-      if Feature.enabled?(:ci_pending_builds_queue_join, runner, default_enabled: :yaml)
-        Ci::Build.joins(:queuing_entry)
+      if Feature.enabled?(:ci_pending_builds_queue_source, runner, default_enabled: :yaml)
+        Ci::PendingBuild.select('build_id AS id').from('ci_pending_builds AS ci_builds')
       else
-        Ci::Build.all
+        Ci::Build.select('id').pending.unstarted
       end
     end
     # rubocop: enable CodeReuse/ActiveRecord
 
+    def running_builds_for_shared_runners
+      all_running_builds
+        .where(runner: Ci::Runner.instance_type)
+        .group(:project_id)
+        .select(:project_id, 'count(*) AS running_builds')
+    end
+
     def new_builds
-      builds = all_builds.pending.unstarted
-      builds = builds.ref_protected if runner.ref_protected?
-      builds
+      if runner.ref_protected?
+        all_builds.where('protected = true')
+        # all_builds.ref_protected
+      else
+        all_builds
+      end
     end
 
     def pre_assign_runner_checks

config/feature_flags/development/ci_pending_builds_queue_join.yml → config/feature_flags/development/ci_pending_builds_queue_source.yml

 ---
-name: ci_pending_builds_queue_join
-introduced_by_url: https://gitlab.com/gitlab-org/gitlab/-/merge_requests/62195
+name: ci_pending_builds_queue_source
+introduced_by_url:
 rollout_issue_url:
-milestone: '13.12'
+milestone: '14.0'
 type: development
 group: group::pipeline execution
 default_enabled: false

ee/app/services/ee/ci/register_job_service.rb

@@ -21,10 +21,10 @@ module EE
 
       # rubocop: disable CodeReuse/ActiveRecord
       def enforce_minutes_based_on_cost_factors(relation)
-        visibility_relation = ::Ci::Build.where(
+        visibility_relation = ::CommitStatus.where(
           projects: { visibility_level: runner.visibility_levels_without_minutes_quota })
 
-        enforce_limits_relation = ::Ci::Build.where('EXISTS (?)', builds_check_limit)
+        enforce_limits_relation = ::CommitStatus.where('EXISTS (?)', builds_check_limit)
 
         relation.merge(visibility_relation.or(enforce_limits_relation))
       end

spec/services/ci/register_job_service_spec.rb

@@ -269,25 +269,25 @@ module Ci
             let!(:unrelated_group_runner) { create(:ci_runner, :group, groups: [unrelated_group]) }
 
             it 'does not consider builds from other group runners' do
-              expect(described_class.new(group_runner).send(:builds_for_group_runner).count).to eq 6
+              expect(described_class.new(group_runner).send(:builds_for_group_runner).size).to eq 6
               execute(group_runner)
 
-              expect(described_class.new(group_runner).send(:builds_for_group_runner).count).to eq 5
+              expect(described_class.new(group_runner).send(:builds_for_group_runner).size).to eq 5
               execute(group_runner)
 
-              expect(described_class.new(group_runner).send(:builds_for_group_runner).count).to eq 4
+              expect(described_class.new(group_runner).send(:builds_for_group_runner).size).to eq 4
               execute(group_runner)
 
-              expect(described_class.new(group_runner).send(:builds_for_group_runner).count).to eq 3
+              expect(described_class.new(group_runner).send(:builds_for_group_runner).size).to eq 3
               execute(group_runner)
 
-              expect(described_class.new(group_runner).send(:builds_for_group_runner).count).to eq 2
+              expect(described_class.new(group_runner).send(:builds_for_group_runner).size).to eq 2
               execute(group_runner)
 
-              expect(described_class.new(group_runner).send(:builds_for_group_runner).count).to eq 1
+              expect(described_class.new(group_runner).send(:builds_for_group_runner).size).to eq 1
               execute(group_runner)
 
-              expect(described_class.new(group_runner).send(:builds_for_group_runner).count).to eq 0
+              expect(described_class.new(group_runner).send(:builds_for_group_runner).size).to eq 0
               expect(execute(group_runner)).to be_nil
             end
           end
@@ -721,17 +721,17 @@ module Ci
         include_examples 'handles runner assignment'
       end
 
-      context 'when joining with pending builds table' do
+      context 'when using pending builds table' do
         before do
-          stub_feature_flags(ci_pending_builds_queue_join: true)
+          stub_feature_flags(ci_pending_builds_queue_source: true)
         end
 
         include_examples 'handles runner assignment'
       end
 
-      context 'when not joining with pending builds table' do
+      context 'when not using pending builds table' do
         before do
-          stub_feature_flags(ci_pending_builds_queue_join: false)
+          stub_feature_flags(ci_pending_builds_queue_source: false)
         end
 
         include_examples 'handles runner assignment'