Add latest changes from gitlab-org/gitlab@16-0-stable-eev16.0.0-rc42

32 files changed, 400 insertions, 160 deletions

diff --git a/app/models/ci/bridge.rb b/app/models/ci/bridge.rb
index 697f06fbffd..7cdd0d56a98 100644
--- a/app/models/ci/bridge.rb
+++ b/app/models/ci/bridge.rb
@@ -55,8 +55,6 @@ module Ci
     end
 
     def retryable?
-      return false unless Feature.enabled?(:ci_recreate_downstream_pipeline, project)
-
       return false if failed? && (pipeline_loop_detected? || reached_max_descendant_pipelines_depth?)
 
       super
@@ -81,7 +79,9 @@ module Ci
       case pipeline.status
       when 'success'
         success!
-      when 'failed', 'canceled', 'skipped'
+      when 'canceled'
+        cancel!
+      when 'failed', 'skipped'
         drop!
       else
         false
diff --git a/app/models/ci/build.rb b/app/models/ci/build.rb
index 1e70dd171ed..61585de4ff7 100644
--- a/app/models/ci/build.rb
+++ b/app/models/ci/build.rb
@@ -18,14 +18,15 @@ module Ci
     belongs_to :runner
     belongs_to :trigger_request
     belongs_to :erased_by, class_name: 'User'
-    belongs_to :pipeline, class_name: 'Ci::Pipeline', foreign_key: :commit_id
+    belongs_to :pipeline, class_name: 'Ci::Pipeline', foreign_key: :commit_id, inverse_of: :builds
 
     RUNNER_FEATURES = {
       upload_multiple_artifacts: -> (build) { build.publishes_artifacts_reports? },
       refspecs: -> (build) { build.merge_request_ref? },
       artifacts_exclude: -> (build) { build.supports_artifacts_exclude? },
       multi_build_steps: -> (build) { build.multi_build_steps? },
-      return_exit_code: -> (build) { build.exit_codes_defined? }
+      return_exit_code: -> (build) { build.exit_codes_defined? },
+      fallback_cache_keys: -> (build) { build.fallback_cache_keys_defined? }
     }.freeze
 
     DEGRADATION_THRESHOLD_VARIABLE_NAME = 'DEGRADATION_THRESHOLD'
@@ -35,8 +36,8 @@ module Ci
 
     has_one :deployment, as: :deployable, class_name: 'Deployment', inverse_of: :deployable
     has_one :pending_state, class_name: 'Ci::BuildPendingState', foreign_key: :build_id, inverse_of: :build
-    has_one :queuing_entry, class_name: 'Ci::PendingBuild', foreign_key: :build_id
-    has_one :runtime_metadata, class_name: 'Ci::RunningBuild', foreign_key: :build_id
+    has_one :queuing_entry, class_name: 'Ci::PendingBuild', foreign_key: :build_id, inverse_of: :build
+    has_one :runtime_metadata, class_name: 'Ci::RunningBuild', foreign_key: :build_id, inverse_of: :build
     has_many :trace_chunks, class_name: 'Ci::BuildTraceChunk', foreign_key: :build_id, inverse_of: :build
     has_many :report_results, class_name: 'Ci::BuildReportResult', foreign_key: :build_id, inverse_of: :build
     has_one :namespace, through: :project
@@ -47,7 +48,7 @@ module Ci
     # Details: https://gitlab.com/gitlab-org/gitlab/-/issues/24644#note_689472685
     has_many :job_artifacts, class_name: 'Ci::JobArtifact', foreign_key: :job_id, dependent: :destroy, inverse_of: :job # rubocop:disable Cop/ActiveRecordDependent
     has_many :job_variables, class_name: 'Ci::JobVariable', foreign_key: :job_id, inverse_of: :job
-    has_many :sourced_pipelines, class_name: 'Ci::Sources::Pipeline', foreign_key: :source_job_id
+    has_many :sourced_pipelines, class_name: 'Ci::Sources::Pipeline', foreign_key: :source_job_id, inverse_of: :build
 
     has_many :pages_deployments, foreign_key: :ci_build_id, inverse_of: :ci_build
 
@@ -55,7 +56,9 @@ module Ci
       has_one :"job_artifacts_#{key}", -> { where(file_type: value) }, class_name: 'Ci::JobArtifact', foreign_key: :job_id, inverse_of: :job
     end
 
-    has_one :runner_machine, through: :metadata, class_name: 'Ci::RunnerMachine'
+    has_one :runner_manager_build, class_name: 'Ci::RunnerManagerBuild', foreign_key: :build_id, inverse_of: :build,
+      autosave: true
+    has_one :runner_manager, foreign_key: :runner_machine_id, through: :runner_manager_build, class_name: 'Ci::RunnerManager'
 
     has_one :runner_session, class_name: 'Ci::BuildRunnerSession', validate: true, foreign_key: :build_id, inverse_of: :build
     has_one :trace_metadata, class_name: 'Ci::BuildTraceMetadata', foreign_key: :build_id, inverse_of: :build
@@ -71,6 +74,7 @@ module Ci
     delegate :gitlab_deploy_token, to: :project
     delegate :harbor_integration, to: :project
     delegate :apple_app_store_integration, to: :project
+    delegate :google_play_integration, to: :project
     delegate :trigger_short_token, to: :trigger_request, allow_nil: true
     delegate :ensure_persistent_ref, to: :pipeline
     delegate :enable_debug_trace!, to: :metadata
@@ -132,7 +136,7 @@ module Ci
 
     scope :eager_load_job_artifacts, -> { includes(:job_artifacts) }
     scope :eager_load_tags, -> { includes(:tags) }
-    scope :eager_load_for_archiving_trace, -> { includes(:project, :pending_state) }
+    scope :eager_load_for_archiving_trace, -> { preload(:project, :pending_state) }
 
     scope :eager_load_everything, -> do
       includes(
@@ -180,7 +184,9 @@ module Ci
 
     acts_as_taggable
 
-    add_authentication_token_field :token, encrypted: :required
+    add_authentication_token_field :token,
+      encrypted: :required,
+      format_with_prefix: :partition_id_prefix_in_16_bit_encode
 
     after_save :stick_build_if_status_changed
 
@@ -592,14 +598,21 @@ module Ci
           .append(key: 'CI_JOB_URL', value: Gitlab::Routing.url_helpers.project_job_url(project, self))
           .append(key: 'CI_JOB_TOKEN', value: token.to_s, public: false, masked: true)
           .append(key: 'CI_JOB_STARTED_AT', value: started_at&.iso8601)
-          .append(key: 'CI_BUILD_ID', value: id.to_s)
-          .append(key: 'CI_BUILD_TOKEN', value: token.to_s, public: false, masked: true)
+
+        if Feature.disabled?(:ci_remove_legacy_predefined_variables, project)
+          variables
+            .append(key: 'CI_BUILD_ID', value: id.to_s)
+            .append(key: 'CI_BUILD_TOKEN', value: token.to_s, public: false, masked: true)
+        end
+
+        variables
           .append(key: 'CI_REGISTRY_USER', value: ::Gitlab::Auth::CI_JOB_USER)
           .append(key: 'CI_REGISTRY_PASSWORD', value: token.to_s, public: false, masked: true)
           .append(key: 'CI_REPOSITORY_URL', value: repo_url.to_s, public: false)
           .concat(deploy_token_variables)
           .concat(harbor_variables)
           .concat(apple_app_store_variables)
+          .concat(google_play_variables)
       end
     end
 
@@ -650,6 +663,13 @@ module Ci
       Gitlab::Ci::Variables::Collection.new(apple_app_store_integration.ci_variables)
     end
 
+    def google_play_variables
+      return [] unless google_play_integration.try(:activated?)
+      return [] unless pipeline.protected_ref?
+
+      Gitlab::Ci::Variables::Collection.new(google_play_integration.ci_variables)
+    end
+
     def features
       {
         trace_sections: true,
@@ -757,9 +777,7 @@ module Ci
     end
 
     def remove_token!
-      if Feature.enabled?(:remove_job_token_on_completion, project)
-        update!(token_encrypted: nil)
-      end
+      update!(token_encrypted: nil)
     end
 
     # acts_as_taggable uses this method create/remove tags with contexts
@@ -802,7 +820,7 @@ module Ci
       return unless project
       return if user&.blocked?
 
-      ActiveRecord::Associations::Preloader.new.preload([self], { runner: :tags })
+      ActiveRecord::Associations::Preloader.new(records: [self], associations: { runner: :tags }).call
 
       project.execute_hooks(build_data.dup, :job_hooks) if project.has_active_hooks?(:job_hooks)
       project.execute_integrations(build_data.dup, :job_hooks) if project.has_active_integrations?(:job_hooks)
@@ -902,9 +920,15 @@ module Ci
     def cache
       cache = Array.wrap(options[:cache])
 
+      cache.each do |single_cache|
+        single_cache[:fallback_keys] = [] unless single_cache.key?(:fallback_keys)
+      end
+
       if project.jobs_cache_index
         cache = cache.map do |single_cache|
-          single_cache.merge(key: "#{single_cache[:key]}-#{project.jobs_cache_index}")
+          cache = single_cache.merge(key: "#{single_cache[:key]}-#{project.jobs_cache_index}")
+          fallback = cache.slice(:fallback_keys).transform_values { |keys| keys.map { |key| "#{key}-#{project.jobs_cache_index}" } }
+          cache.merge(fallback.compact)
         end
       end
 
@@ -913,10 +937,16 @@ module Ci
       cache.map do |entry|
         type_suffix = !entry[:unprotect] && pipeline.protected_ref? ? 'protected' : 'non_protected'
 
-        entry.merge(key: "#{entry[:key]}-#{type_suffix}")
+        cache = entry.merge(key: "#{entry[:key]}-#{type_suffix}")
+        fallback = cache.slice(:fallback_keys).transform_values { |keys| keys.map { |key| "#{key}-#{type_suffix}" } }
+        cache.merge(fallback.compact)
       end
     end
 
+    def fallback_cache_keys_defined?
+      Array.wrap(options[:cache]).any? { |cache| cache[:fallback_keys].present? }
+    end
+
     def credentials
       Gitlab::Ci::Build::Credentials::Factory.new(self).create!
     end
@@ -1091,10 +1121,6 @@ module Ci
       ::Ci::PendingBuild.upsert_from_build!(self)
     end
 
-    def create_runtime_metadata!
-      ::Ci::RunningBuild.upsert_shared_runner_build!(self)
-    end
-
     ##
     # We can have only one queuing entry or running build tracking entry,
     # because there is a unique index on `build_id` in each table, but we need
@@ -1161,11 +1187,6 @@ module Ci
       end
     end
 
-    override :format_token
-    def format_token(token)
-      "#{partition_id.to_s(16)}_#{token}"
-    end
-
     protected
 
     def run_status_commit_hooks!
@@ -1231,10 +1252,10 @@ module Ci
     end
 
     def job_jwt_variables
-      if project.ci_cd_settings.opt_in_jwt?
+      if id_tokens?
         id_tokens_variables
       else
-        predefined_jwt_variables.concat(id_tokens_variables)
+        predefined_jwt_variables
       end
     end
 
@@ -1251,8 +1272,6 @@ module Ci
     end
 
     def id_tokens_variables
-      return [] unless id_tokens?
-
       Gitlab::Ci::Variables::Collection.new.tap do |variables|
         id_tokens.each do |var_name, token_data|
           token = Gitlab::Ci::JwtV2.for_build(self, aud: token_data['aud'])
@@ -1308,6 +1327,10 @@ module Ci
         ).to_context]
       )
     end
+
+    def partition_id_prefix_in_16_bit_encode
+      "#{partition_id.to_s(16)}_"
+    end
   end
 end
 
diff --git a/app/models/ci/build_metadata.rb b/app/models/ci/build_metadata.rb
index b294afd405d..382f861a802 100644
--- a/app/models/ci/build_metadata.rb
+++ b/app/models/ci/build_metadata.rb
@@ -10,15 +10,16 @@ module Ci
     include Presentable
     include ChronicDurationAttribute
     include Gitlab::Utils::StrongMemoize
+    include SafelyChangeColumnDefault
 
     self.table_name = 'p_ci_builds_metadata'
     self.primary_key = 'id'
+    columns_changing_default :partition_id
 
     partitionable scope: :build
 
     belongs_to :build, class_name: 'CommitStatus'
     belongs_to :project
-    belongs_to :runner_machine, class_name: 'Ci::RunnerMachine'
 
     before_create :set_build_project
 
diff --git a/app/models/ci/build_need.rb b/app/models/ci/build_need.rb
index 03d1bd14bfb..940221619b3 100644
--- a/app/models/ci/build_need.rb
+++ b/app/models/ci/build_need.rb
@@ -6,8 +6,6 @@ module Ci
     include BulkInsertSafe
     include IgnorableColumns
 
-    ignore_column :id_convert_to_bigint, remove_with: '16.0', remove_after: '2023-04-22'
-
     belongs_to :build, class_name: "Ci::Processable", foreign_key: :build_id, inverse_of: :needs
 
     partitionable scope: :build
diff --git a/app/models/ci/build_pending_state.rb b/app/models/ci/build_pending_state.rb
index 3684dac06c7..966884ae158 100644
--- a/app/models/ci/build_pending_state.rb
+++ b/app/models/ci/build_pending_state.rb
@@ -3,7 +3,7 @@
 class Ci::BuildPendingState < Ci::ApplicationRecord
   include Ci::Partitionable
 
-  belongs_to :build, class_name: 'Ci::Build', foreign_key: :build_id
+  belongs_to :build, class_name: 'Ci::Build', foreign_key: :build_id, inverse_of: :pending_state
 
   partitionable scope: :build
 
diff --git a/app/models/ci/build_trace_chunk.rb b/app/models/ci/build_trace_chunk.rb
index 541a8b5bffa..03b59b19ef1 100644
--- a/app/models/ci/build_trace_chunk.rb
+++ b/app/models/ci/build_trace_chunk.rb
@@ -9,7 +9,7 @@ module Ci
     include ::Gitlab::ExclusiveLeaseHelpers
     include ::Gitlab::OptimisticLocking
 
-    belongs_to :build, class_name: "Ci::Build", foreign_key: :build_id
+    belongs_to :build, class_name: "Ci::Build", foreign_key: :build_id, inverse_of: :trace_chunks
 
     partitionable scope: :build
 
diff --git a/app/models/ci/build_trace_metadata.rb b/app/models/ci/build_trace_metadata.rb
index 00cf1531483..4c76089617f 100644
--- a/app/models/ci/build_trace_metadata.rb
+++ b/app/models/ci/build_trace_metadata.rb
@@ -42,9 +42,7 @@ module Ci
     end
 
     def track_archival!(trace_artifact_id, checksum)
-      update!(trace_artifact_id: trace_artifact_id,
-              checksum: checksum,
-              archived_at: Time.current)
+      update!(trace_artifact_id: trace_artifact_id, checksum: checksum, archived_at: Time.current)
     end
 
     def archival_attempts_message
diff --git a/app/models/ci/catalog/listing.rb b/app/models/ci/catalog/listing.rb
new file mode 100644
index 00000000000..b9e777f27a0
--- /dev/null
+++ b/app/models/ci/catalog/listing.rb
@@ -0,0 +1,34 @@
+# frozen_string_literal: true
+
+module Ci
+  module Catalog
+    class Listing
+      # This class is the SSoT to displaying the list of resources in the
+      # CI/CD Catalog given a namespace as a scope.
+      # This model is not directly backed by a table and joins catalog resources
+      # with projects to return relevant data.
+      def initialize(namespace, current_user)
+        raise ArgumentError, 'Namespace is not a root namespace' unless namespace.root?
+
+        @namespace = namespace
+        @current_user = current_user
+      end
+
+      def resources
+        Ci::Catalog::Resource
+          .joins(:project).includes(:project)
+          .merge(projects_in_namespace_visible_to_user)
+      end
+
+      private
+
+      attr_reader :namespace, :current_user
+
+      def projects_in_namespace_visible_to_user
+        Project
+          .in_namespace(namespace.self_and_descendant_ids)
+          .public_or_visible_to_user(current_user, ::Gitlab::Access::DEVELOPER)
+      end
+    end
+  end
+end
diff --git a/app/models/ci/catalog/resource.rb b/app/models/ci/catalog/resource.rb
new file mode 100644
index 00000000000..bb4584aacae
--- /dev/null
+++ b/app/models/ci/catalog/resource.rb
@@ -0,0 +1,28 @@
+# frozen_string_literal: true
+
+module Ci
+  module Catalog
+    # This class represents a CI/CD Catalog resource.
+    # A Catalog resource is normally associated to a project.
+    # This model connects to the `main` database because of its
+    # dependency on the Project model and its need to join with that table
+    # in order to generate the CI/CD catalog.
+    class Resource < ::ApplicationRecord
+      self.table_name = 'catalog_resources'
+
+      belongs_to :project
+
+      scope :for_projects, ->(project_ids) { where(project_id: project_ids) }
+
+      delegate :avatar_path, :description, :name, to: :project
+
+      def versions
+        project.releases.order_released_desc
+      end
+
+      def latest_version
+        versions.first
+      end
+    end
+  end
+end
diff --git a/app/models/ci/commit_with_pipeline.rb b/app/models/ci/commit_with_pipeline.rb
index dde4b534aaa..2aa249df321 100644
--- a/app/models/ci/commit_with_pipeline.rb
+++ b/app/models/ci/commit_with_pipeline.rb
@@ -19,7 +19,7 @@ class Ci::CommitWithPipeline < SimpleDelegator
   end
 
   def lazy_latest_pipeline
-    BatchLoader.for(sha).batch do |shas, loader|
+    BatchLoader.for(sha).batch(key: project.id) do |shas, loader|
       preload_pipelines = project.ci_pipelines.latest_pipeline_per_commit(shas.compact)
 
       shas.each do |sha|
diff --git a/app/models/ci/daily_build_group_report_result.rb b/app/models/ci/daily_build_group_report_result.rb
index 598d1456a48..5ec54ee2983 100644
--- a/app/models/ci/daily_build_group_report_result.rb
+++ b/app/models/ci/daily_build_group_report_result.rb
@@ -4,9 +4,10 @@ module Ci
   class DailyBuildGroupReportResult < Ci::ApplicationRecord
     PARAM_TYPES = %w[coverage].freeze
 
-    belongs_to :last_pipeline, class_name: 'Ci::Pipeline', foreign_key: :last_pipeline_id
+    belongs_to :last_pipeline, class_name: 'Ci::Pipeline', foreign_key: :last_pipeline_id,
+      inverse_of: :daily_build_group_report_results
     belongs_to :project
-    belongs_to :group
+    belongs_to :group, class_name: '::Group'
 
     validates :data, json_schema: { filename: "daily_build_group_report_result_data" }
 
diff --git a/app/models/ci/group_variable.rb b/app/models/ci/group_variable.rb
index b03c46a164f..f04f0d27e51 100644
--- a/app/models/ci/group_variable.rb
+++ b/app/models/ci/group_variable.rb
@@ -29,5 +29,13 @@ module Ci
     def audit_details
       key
     end
+
+    def group_name
+      group.name
+    end
+
+    def group_ci_cd_settings_path
+      Gitlab::Routing.url_helpers.group_settings_ci_cd_path(group)
+    end
   end
 end
diff --git a/app/models/ci/job_artifact.rb b/app/models/ci/job_artifact.rb
index 89a3d269a43..766155c6a99 100644
--- a/app/models/ci/job_artifact.rb
+++ b/app/models/ci/job_artifact.rb
@@ -132,7 +132,7 @@ module Ci
     PLAN_LIMIT_PREFIX = 'ci_max_artifact_size_'
 
     belongs_to :project
-    belongs_to :job, class_name: "Ci::Build", foreign_key: :job_id
+    belongs_to :job, class_name: "Ci::Build", foreign_key: :job_id, inverse_of: :job_artifacts
 
     mount_file_store_uploader JobArtifactUploader, skip_store_file: true
 
@@ -155,7 +155,7 @@ module Ci
     scope :not_expired, -> { where('expire_at IS NULL OR expire_at > ?', Time.current) }
     scope :for_sha, ->(sha, project_id) { joins(job: :pipeline).where(ci_pipelines: { sha: sha, project_id: project_id }) }
     scope :for_job_ids, ->(job_ids) { where(job_id: job_ids) }
-    scope :for_job_name, ->(name) { joins(:job).where(ci_builds: { name: name }) }
+    scope :for_job_name, ->(name) { joins(:job).merge(Ci::Build.by_name(name)) }
     scope :created_at_before, ->(time) { where(arel_table[:created_at].lteq(time)) }
     scope :id_before, ->(id) { where(arel_table[:id].lteq(id)) }
     scope :id_after, ->(id) { where(arel_table[:id].gt(id)) }
@@ -177,6 +177,8 @@ module Ci
       where(file_type: self.erasable_file_types)
     end
 
+    scope :non_trace, -> { where.not(file_type: [:trace]) }
+
     scope :downloadable, -> { where(file_type: DOWNLOADABLE_TYPES) }
     scope :unlocked, -> { joins(job: :pipeline).merge(::Ci::Pipeline.unlocked) }
     scope :order_expired_asc, -> { order(expire_at: :asc) }
diff --git a/app/models/ci/job_token/scope.rb b/app/models/ci/job_token/scope.rb
index 20775077bd8..f389c642fd8 100644
--- a/app/models/ci/job_token/scope.rb
+++ b/app/models/ci/job_token/scope.rb
@@ -58,8 +58,7 @@ module Ci
       end
 
       def inbound_accessible?(accessed_project)
-        # if the flag or setting is disabled any project is considered to be in scope.
-        return true unless Feature.enabled?(:ci_inbound_job_token_scope, accessed_project)
+        # if the setting is disabled any project is considered to be in scope.
         return true unless accessed_project.ci_inbound_job_token_scope_enabled?
 
         inbound_linked_as_accessible?(accessed_project)
diff --git a/app/models/ci/job_variable.rb b/app/models/ci/job_variable.rb
index 998f0647ad5..573999995bc 100644
--- a/app/models/ci/job_variable.rb
+++ b/app/models/ci/job_variable.rb
@@ -7,7 +7,7 @@ module Ci
     include Ci::RawVariable
     include BulkInsertSafe
 
-    belongs_to :job, class_name: "Ci::Build", foreign_key: :job_id
+    belongs_to :job, class_name: "Ci::Build", foreign_key: :job_id, inverse_of: :job_variables
 
     partitionable scope: :job
 
diff --git a/app/models/ci/namespace_mirror.rb b/app/models/ci/namespace_mirror.rb
index 5ea51fbe0a7..ff7e681217a 100644
--- a/app/models/ci/namespace_mirror.rb
+++ b/app/models/ci/namespace_mirror.rb
@@ -41,8 +41,7 @@ module Ci
         namespace = event.namespace
         traversal_ids = namespace.self_and_ancestor_ids(hierarchy_order: :desc)
 
-        upsert({ namespace_id: event.namespace_id, traversal_ids: traversal_ids },
-               unique_by: :namespace_id)
+        upsert({ namespace_id: event.namespace_id, traversal_ids: traversal_ids }, unique_by: :namespace_id)
       end
     end
   end
diff --git a/app/models/ci/pending_build.rb b/app/models/ci/pending_build.rb
index 2b1eb67d4f2..14050a1e78e 100644
--- a/app/models/ci/pending_build.rb
+++ b/app/models/ci/pending_build.rb
@@ -14,7 +14,6 @@ module Ci
     validates :namespace, presence: true
 
     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) }
     scope :for_tags, ->(tag_ids) do
       if tag_ids.present?
diff --git a/app/models/ci/pipeline.rb b/app/models/ci/pipeline.rb
index bd426e02b9c..babea831d85 100644
--- a/app/models/ci/pipeline.rb
+++ b/app/models/ci/pipeline.rb
@@ -11,7 +11,6 @@ module Ci
     include Gitlab::OptimisticLocking
     include Gitlab::Utils::StrongMemoize
     include AtomicInternalId
-    include EnumWithNil
     include Ci::HasRef
     include ShaAttribute
     include FromUnion
@@ -19,6 +18,9 @@ module Ci
     include EachBatch
     include FastDestroyAll::Helpers
 
+    include IgnorableColumns
+    ignore_column :id_convert_to_bigint, remove_with: '16.3', remove_after: '2023-08-22'
+
     MAX_OPEN_MERGE_REQUESTS_REFS = 4
 
     PROJECT_ROUTE_AND_NAMESPACE_ROUTE = {
@@ -46,39 +48,53 @@ module Ci
 
     belongs_to :project, inverse_of: :all_pipelines
     belongs_to :user
-    belongs_to :auto_canceled_by, class_name: 'Ci::Pipeline'
+    belongs_to :auto_canceled_by, class_name: 'Ci::Pipeline', inverse_of: :auto_canceled_pipelines
     belongs_to :pipeline_schedule, class_name: 'Ci::PipelineSchedule'
     belongs_to :merge_request, class_name: 'MergeRequest'
     belongs_to :external_pull_request
     belongs_to :ci_ref, class_name: 'Ci::Ref', foreign_key: :ci_ref_id, inverse_of: :pipelines
 
     has_internal_id :iid, scope: :project, presence: false,
-                          track_if: -> { !importing? },
-                          ensure_if: -> { !importing? },
-                          init: ->(pipeline, scope) do
-                                  if pipeline
-                                    pipeline.project&.all_pipelines&.maximum(:iid) || pipeline.project&.all_pipelines&.count
-                                  elsif scope
-                                    ::Ci::Pipeline.where(**scope).maximum(:iid)
-                                  end
-                                end
+      track_if: -> { !importing? },
+      ensure_if: -> { !importing? },
+      init: ->(pipeline, scope) do
+        if pipeline
+          pipeline.project&.all_pipelines&.maximum(:iid) || pipeline.project&.all_pipelines&.count
+        elsif scope
+          ::Ci::Pipeline.where(**scope).maximum(:iid)
+        end
+      end
 
     has_many :stages, -> { order(position: :asc) }, inverse_of: :pipeline
+
+    #
+    # In https://gitlab.com/groups/gitlab-org/-/epics/9991, we aim to convert all CommitStatus related models to
+    # Ci:Job models. With that epic, we aim to replace `statuses` with `jobs`.
+    #
+    # DEPRECATED:
     has_many :statuses, class_name: 'CommitStatus', foreign_key: :commit_id, inverse_of: :pipeline
+    has_many :processables, class_name: 'Ci::Processable', foreign_key: :commit_id, inverse_of: :pipeline
     has_many :latest_statuses_ordered_by_stage, -> { latest.order(:stage_idx, :stage) }, class_name: 'CommitStatus', foreign_key: :commit_id, inverse_of: :pipeline
     has_many :latest_statuses, -> { latest }, class_name: 'CommitStatus', foreign_key: :commit_id, inverse_of: :pipeline
-    has_many :statuses_order_id_desc, -> { order_id_desc }, class_name: 'CommitStatus', foreign_key: :commit_id
-    has_many :processables, class_name: 'Ci::Processable', foreign_key: :commit_id, inverse_of: :pipeline
+    has_many :statuses_order_id_desc, -> { order_id_desc }, class_name: 'CommitStatus', foreign_key: :commit_id,
+      inverse_of: :pipeline
     has_many :bridges, class_name: 'Ci::Bridge', foreign_key: :commit_id, inverse_of: :pipeline
     has_many :builds, foreign_key: :commit_id, inverse_of: :pipeline
     has_many :generic_commit_statuses, foreign_key: :commit_id, inverse_of: :pipeline, class_name: 'GenericCommitStatus'
+    #
+    # NEW:
+    has_many :all_jobs, class_name: 'CommitStatus', foreign_key: :commit_id, inverse_of: :pipeline
+    has_many :current_jobs, -> { latest }, class_name: 'CommitStatus', foreign_key: :commit_id, inverse_of: :pipeline
+    has_many :all_processable_jobs, class_name: 'Ci::Processable', foreign_key: :commit_id, inverse_of: :pipeline
+    has_many :current_processable_jobs, -> { latest }, class_name: 'Ci::Processable', foreign_key: :commit_id, inverse_of: :pipeline
+
     has_many :job_artifacts, through: :builds
     has_many :build_trace_chunks, class_name: 'Ci::BuildTraceChunk', through: :builds, source: :trace_chunks
-    has_many :trigger_requests, dependent: :destroy, foreign_key: :commit_id # rubocop:disable Cop/ActiveRecordDependent
+    has_many :trigger_requests, dependent: :destroy, foreign_key: :commit_id, inverse_of: :pipeline # rubocop:disable Cop/ActiveRecordDependent
     has_many :variables, class_name: 'Ci::PipelineVariable'
     has_many :latest_builds, -> { latest.with_project_and_metadata }, foreign_key: :commit_id, inverse_of: :pipeline, class_name: 'Ci::Build'
     has_many :downloadable_artifacts, -> do
-      not_expired.or(where_exists(::Ci::Pipeline.artifacts_locked.where('ci_pipelines.id = ci_builds.commit_id'))).downloadable.with_job
+      not_expired.or(where_exists(Ci::Pipeline.artifacts_locked.where("#{Ci::Pipeline.quoted_table_name}.id = #{Ci::Build.quoted_table_name}.commit_id"))).downloadable.with_job
     end, through: :latest_builds, source: :job_artifacts
     has_many :latest_successful_builds, -> { latest.success.with_project_and_metadata }, foreign_key: :commit_id, inverse_of: :pipeline, class_name: 'Ci::Build'
 
@@ -86,17 +102,24 @@ module Ci
 
     # Merge requests for which the current pipeline is running against
     # the merge request's latest commit.
-    has_many :merge_requests_as_head_pipeline, foreign_key: "head_pipeline_id", class_name: 'MergeRequest'
+    has_many :merge_requests_as_head_pipeline, foreign_key: :head_pipeline_id, class_name: 'MergeRequest',
+      inverse_of: :head_pipeline
+
     has_many :pending_builds, -> { pending }, foreign_key: :commit_id, class_name: 'Ci::Build', inverse_of: :pipeline
-    has_many :failed_builds, -> { latest.failed }, foreign_key: :commit_id, class_name: 'Ci::Build', inverse_of: :pipeline
+    has_many :failed_builds, -> { latest.failed }, foreign_key: :commit_id, class_name: 'Ci::Build',
+      inverse_of: :pipeline
     has_many :retryable_builds, -> { latest.failed_or_canceled.includes(:project) }, foreign_key: :commit_id, class_name: 'Ci::Build', inverse_of: :pipeline
-    has_many :cancelable_statuses, -> { cancelable }, foreign_key: :commit_id, class_name: 'CommitStatus'
+    has_many :cancelable_statuses, -> { cancelable }, foreign_key: :commit_id, class_name: 'CommitStatus',
+      inverse_of: :pipeline
     has_many :manual_actions, -> { latest.manual_actions.includes(:project) }, foreign_key: :commit_id, class_name: 'Ci::Build', inverse_of: :pipeline
     has_many :scheduled_actions, -> { latest.scheduled_actions.includes(:project) }, foreign_key: :commit_id, class_name: 'Ci::Build', inverse_of: :pipeline
 
-    has_many :auto_canceled_pipelines, class_name: 'Ci::Pipeline', foreign_key: 'auto_canceled_by_id'
-    has_many :auto_canceled_jobs, class_name: 'CommitStatus', foreign_key: 'auto_canceled_by_id'
-    has_many :sourced_pipelines, class_name: 'Ci::Sources::Pipeline', foreign_key: :source_pipeline_id
+    has_many :auto_canceled_pipelines, class_name: 'Ci::Pipeline', foreign_key: :auto_canceled_by_id,
+      inverse_of: :auto_canceled_by
+    has_many :auto_canceled_jobs, class_name: 'CommitStatus', foreign_key: :auto_canceled_by_id,
+      inverse_of: :auto_canceled_by
+    has_many :sourced_pipelines, class_name: 'Ci::Sources::Pipeline', foreign_key: :source_pipeline_id,
+      inverse_of: :source_pipeline
 
     has_one :source_pipeline, class_name: 'Ci::Sources::Pipeline', inverse_of: :pipeline
 
@@ -114,7 +137,9 @@ module Ci
 
     has_one :pipeline_metadata, class_name: 'Ci::PipelineMetadata', inverse_of: :pipeline
 
-    has_many :daily_build_group_report_results, class_name: 'Ci::DailyBuildGroupReportResult', foreign_key: :last_pipeline_id
+    has_many :daily_build_group_report_results, class_name: 'Ci::DailyBuildGroupReportResult',
+      foreign_key: :last_pipeline_id, inverse_of: :last_pipeline
+
     has_many :latest_builds_report_results, through: :latest_builds, source: :report_results
     has_many :pipeline_artifacts, class_name: 'Ci::PipelineArtifact', inverse_of: :pipeline, dependent: :destroy # rubocop:disable Cop/ActiveRecordDependent
 
@@ -143,9 +168,9 @@ module Ci
 
     # We use `Enums::Ci::Pipeline.sources` here so that EE can more easily extend
     # this `Hash` with new values.
-    enum_with_nil source: Enums::Ci::Pipeline.sources
+    enum source: Enums::Ci::Pipeline.sources
 
-    enum_with_nil config_source: Enums::Ci::Pipeline.config_sources
+    enum config_source: Enums::Ci::Pipeline.config_sources
 
     # We use `Enums::Ci::Pipeline.failure_reasons` here so that EE can more easily
     # extend this `Hash` with new values.
@@ -336,6 +361,22 @@ module Ci
           AutoDevops::DisableWorker.perform_async(pipeline.id) if pipeline.auto_devops_source?
         end
       end
+
+      after_transition any => [:running, *::Ci::Pipeline.completed_statuses] do |pipeline|
+        project = pipeline&.project
+
+        next unless project
+        next unless Feature.enabled?(:pipeline_trigger_merge_status, project)
+
+        pipeline.run_after_commit do
+          next if pipeline.child?
+          next unless project.only_allow_merge_if_pipeline_succeeds?(inherit_group_setting: true)
+
+          pipeline.all_merge_requests.opened.each do |merge_request|
+            GraphqlTriggers.merge_request_merge_status_updated(merge_request)
+          end
+        end
+      end
     end
 
     scope :internal, -> { where(source: internal_sources) }
@@ -361,18 +402,25 @@ module Ci
     scope :created_before_id, -> (id) { where(arel_table[:id].lt(id)) }
     scope :before_pipeline, -> (pipeline) { created_before_id(pipeline.id).outside_pipeline_family(pipeline) }
     scope :with_pipeline_source, -> (source) { where(source: source) }
+    scope :preload_pipeline_metadata, -> { preload(:pipeline_metadata) }
 
     scope :outside_pipeline_family, ->(pipeline) do
       where.not(id: pipeline.same_family_pipeline_ids)
     end
 
     scope :with_reports, -> (reports_scope) do
-      where('EXISTS (?)', ::Ci::Build.latest.with_artifacts(reports_scope).where('ci_pipelines.id=ci_builds.commit_id').select(1))
+      where('EXISTS (?)',
+        ::Ci::Build
+          .latest
+          .with_artifacts(reports_scope)
+          .where("#{quoted_table_name}.id = #{Ci::Build.quoted_table_name}.commit_id")
+          .select(1)
+      )
     end
 
     scope :with_only_interruptible_builds, -> do
       where('NOT EXISTS (?)',
-        Ci::Build.where('ci_builds.commit_id = ci_pipelines.id')
+        Ci::Build.where("#{Ci::Build.quoted_table_name}.commit_id = #{quoted_table_name}.id")
                  .with_status(STARTED_STATUSES)
                  .not_interruptible
       )
@@ -382,11 +430,15 @@ module Ci
     # In general, please use `Ci::PipelinesForMergeRequestFinder` instead,
     # for checking permission of the actor.
     scope :triggered_by_merge_request, -> (merge_request) do
-      where(source: :merge_request_event,
-            merge_request: merge_request,
-            project: [merge_request.source_project, merge_request.target_project])
+      where(
+        source: :merge_request_event,
+        merge_request: merge_request,
+        project: [merge_request.source_project, merge_request.target_project]
+      )
     end
 
+    scope :order_id_desc, -> { order(id: :desc) }
+
     # Returns the pipelines in descending order (= newest first), optionally
     # limited to a number of references.
     #
@@ -657,7 +709,7 @@ module Ci
     # rubocop: enable CodeReuse/ServiceClass
 
     def lazy_ref_commit
-      BatchLoader.for(ref).batch do |refs, loader|
+      BatchLoader.for(ref).batch(key: project.id) do |refs, loader|
         next unless project.repository_exists?
 
         project.repository.list_commits_by_ref_name(refs).then do |commits|
@@ -818,8 +870,7 @@ module Ci
         when 'manual' then block
         when 'scheduled' then delay
         else
-          raise Ci::HasStatus::UnknownStatusError,
-                "Unknown status `#{new_status}`"
+          raise Ci::HasStatus::UnknownStatusError, "Unknown status `#{new_status}`"
         end
       end
     end
@@ -1282,7 +1333,7 @@ module Ci
       types_to_collect = report_types.empty? ? ::Ci::JobArtifact::SECURITY_REPORT_FILE_TYPES : report_types
 
       ::Gitlab::Ci::Reports::Security::Reports.new(self).tap do |security_reports|
-        latest_report_builds(reports_scope).each do |build|
+        latest_report_builds_in_self_and_project_descendants(reports_scope).includes(pipeline: { project: :route }).each do |build| # rubocop:disable Rails/FindEach
           build.collect_security_reports!(security_reports, report_types: types_to_collect)
         end
       end
@@ -1294,7 +1345,7 @@ module Ci
 
     def cluster_agent_authorizations
       strong_memoize(:cluster_agent_authorizations) do
-        ::Clusters::AgentAuthorizationsFinder.new(project).execute
+        ::Clusters::Agents::Authorizations::CiAccess::Finder.new(project).execute
       end
     end
 
diff --git a/app/models/ci/pipeline_schedule.rb b/app/models/ci/pipeline_schedule.rb
index 20ff07e88ba..49d27053745 100644
--- a/app/models/ci/pipeline_schedule.rb
+++ b/app/models/ci/pipeline_schedule.rb
@@ -8,14 +8,15 @@ module Ci
     include CronSchedulable
     include Limitable
     include EachBatch
+    include BatchNullifyDependentAssociations
 
     self.limit_name = 'ci_pipeline_schedules'
     self.limit_scope = :project
 
     belongs_to :project
     belongs_to :owner, class_name: 'User'
-    has_one :last_pipeline, -> { order(id: :desc) }, class_name: 'Ci::Pipeline'
-    has_many :pipelines
+    has_one :last_pipeline, -> { order(id: :desc) }, class_name: 'Ci::Pipeline', inverse_of: :pipeline_schedule
+    has_many :pipelines, dependent: :nullify # rubocop:disable Cop/ActiveRecordDependent
     has_many :variables, class_name: 'Ci::PipelineScheduleVariable'
 
     validates :cron, unless: :importing?, cron: true, presence: { unless: :importing? }
@@ -81,6 +82,14 @@ module Ci
     def worker_cron_expression
       Settings.cron_jobs['pipeline_schedule_worker']['cron']
     end
+
+    # Using destroy instead of before_destroy as we want nullify_dependent_associations_in_batches
+    # to run first and not in a transaction block. This prevents timeouts for schedules with numerous pipelines
+    def destroy
+      nullify_dependent_associations_in_batches
+
+      super
+    end
   end
 end
 
diff --git a/app/models/ci/pipeline_variable.rb b/app/models/ci/pipeline_variable.rb
index 8e83b41cd0b..f2457af0074 100644
--- a/app/models/ci/pipeline_variable.rb
+++ b/app/models/ci/pipeline_variable.rb
@@ -6,6 +6,9 @@ module Ci
     include Ci::HasVariable
     include Ci::RawVariable
 
+    include IgnorableColumns
+    ignore_column :id_convert_to_bigint, remove_with: '16.3', remove_after: '2023-08-22'
+
     belongs_to :pipeline
 
     partitionable scope: :pipeline
diff --git a/app/models/ci/processable.rb b/app/models/ci/processable.rb
index 37c82c125aa..4c421f066f9 100644
--- a/app/models/ci/processable.rb
+++ b/app/models/ci/processable.rb
@@ -1,6 +1,8 @@
 # frozen_string_literal: true
 
 module Ci
+  # This class is a collection of common features between Ci::Build and Ci::Bridge.
+  # In https://gitlab.com/groups/gitlab-org/-/epics/9991, we aim to clarify class naming conventions.
   class Processable < ::CommitStatus
     include Gitlab::Utils::StrongMemoize
     include FromUnion
diff --git a/app/models/ci/project_mirror.rb b/app/models/ci/project_mirror.rb
index 15a161d5b7c..23cd5d92730 100644
--- a/app/models/ci/project_mirror.rb
+++ b/app/models/ci/project_mirror.rb
@@ -13,8 +13,7 @@ module Ci
 
     class << self
       def sync!(event)
-        upsert({ project_id: event.project_id, namespace_id: event.project.namespace_id },
-               unique_by: :project_id)
+        upsert({ project_id: event.project_id, namespace_id: event.project.namespace_id }, unique_by: :project_id)
       end
     end
   end
diff --git a/app/models/ci/ref.rb b/app/models/ci/ref.rb
index af5fdabff6e..199e1cd07e7 100644
--- a/app/models/ci/ref.rb
+++ b/app/models/ci/ref.rb
@@ -43,8 +43,7 @@ module Ci
 
     class << self
       def ensure_for(pipeline)
-        safe_find_or_create_by(project_id: pipeline.project_id,
-                               ref_path: pipeline.source_ref_path)
+        safe_find_or_create_by(project_id: pipeline.project_id, ref_path: pipeline.source_ref_path)
       end
 
       def failing_state?(status_name)
diff --git a/app/models/ci/resource_group.rb b/app/models/ci/resource_group.rb
index b788e4f58c1..48f321a236d 100644
--- a/app/models/ci/resource_group.rb
+++ b/app/models/ci/resource_group.rb
@@ -29,13 +29,19 @@ module Ci
         partition_id: processable.partition_id
       }
 
-      resources.free.limit(1).update_all(attrs) > 0
+      success = resources.free.limit(1).update_all(attrs) > 0
+      log_event(success: success, processable: processable, action: "assign resource to processable")
+
+      success
     end
 
     def release_resource_from(processable)
       attrs = { build_id: nil, partition_id: nil }
 
-      resources.retained_by(processable).update_all(attrs) > 0
+      success = resources.retained_by(processable).update_all(attrs) > 0
+      log_event(success: success, processable: processable, action: "release resource from processable")
+
+      success
     end
 
     def upcoming_processables
@@ -52,6 +58,10 @@ module Ci
       end
     end
 
+    def current_processable
+      Ci::Processable.find_by('(id, partition_id) IN (?)', resources.select('build_id, partition_id'))
+    end
+
     private
 
     # In order to avoid deadlock, we do NOT specify the job execution order in the same pipeline.
@@ -72,5 +82,14 @@ module Ci
       # belong to the same resource group are executed once at time.
       self.resources.build if self.resources.empty?
     end
+
+    def log_event(success:, processable:, action:)
+      Gitlab::Ci::ResourceGroups::Logger.build.info({
+        resource_group_id: self.id,
+        processable_id: processable.id,
+        message: "attempted to #{action}",
+        success: success
+      })
+    end
   end
 end
diff --git a/app/models/ci/runner.rb b/app/models/ci/runner.rb
index 09ac0fa69e7..7727e94875b 100644
--- a/app/models/ci/runner.rb
+++ b/app/models/ci/runner.rb
@@ -17,7 +17,10 @@ module Ci
 
     extend ::Gitlab::Utils::Override
 
-    add_authentication_token_field :token, encrypted: :optional, expires_at: :compute_token_expiration
+    add_authentication_token_field :token,
+      encrypted: :optional,
+      expires_at: :compute_token_expiration,
+      format_with_prefix: :prefix_for_new_and_legacy_runner
 
     enum access_level: {
       not_protected: 0,
@@ -54,6 +57,9 @@ module Ci
     # The `STALE_TIMEOUT` constant defines the how far past the last contact or creation date a runner will be considered stale
     STALE_TIMEOUT = 3.months
 
+    # Only allow authentication token to be visible for a short while
+    REGISTRATION_AVAILABILITY_TIME = 1.hour
+
     AVAILABLE_TYPES_LEGACY = %w[specific shared].freeze
     AVAILABLE_TYPES = runner_types.keys.freeze
     AVAILABLE_STATUSES = %w[active paused online offline never_contacted stale].freeze # TODO: Remove in %16.0: active, paused. Relevant issue: https://gitlab.com/gitlab-org/gitlab/-/issues/344648
@@ -64,7 +70,7 @@ module Ci
 
     TAG_LIST_MAX_LENGTH = 50
 
-    has_many :runner_machines, inverse_of: :runner
+    has_many :runner_managers, inverse_of: :runner
     has_many :builds
     has_many :runner_projects, inverse_of: :runner, autosave: true, dependent: :destroy # rubocop:disable Cop/ActiveRecordDependent
     has_many :projects, through: :runner_projects, disable_joins: true
@@ -81,8 +87,13 @@ module Ci
     scope :active, -> (value = true) { where(active: value) }
     scope :paused, -> { active(false) }
     scope :online, -> { where('contacted_at > ?', online_contact_time_deadline) }
-    scope :recent, -> { where('ci_runners.created_at >= :date OR ci_runners.contacted_at >= :date', date: stale_deadline) }
-    scope :stale, -> { where('ci_runners.created_at < :date AND (ci_runners.contacted_at IS NULL OR ci_runners.contacted_at < :date)', date: stale_deadline) }
+    scope :recent, -> do
+      where('ci_runners.created_at >= :datetime OR ci_runners.contacted_at >= :datetime', datetime: stale_deadline)
+    end
+    scope :stale, -> do
+      where('ci_runners.created_at <= :datetime AND ' \
+            '(ci_runners.contacted_at IS NULL OR ci_runners.contacted_at <= :datetime)', datetime: stale_deadline)
+    end
     scope :offline, -> { where(arel_table[:contacted_at].lteq(online_contact_time_deadline)) }
     scope :never_contacted, -> { where(contacted_at: nil) }
     scope :ordered, -> { order(id: :desc) }
@@ -123,7 +134,7 @@ module Ci
       belonging_to_group(group_self_and_ancestors_ids)
     }
 
-    scope :belonging_to_parent_group_of_project, -> (project_id) {
+    scope :belonging_to_parent_groups_of_project, -> (project_id) {
       raise ArgumentError, "only 1 project_id allowed for performance reasons" unless project_id.is_a?(Integer)
 
       project_groups = ::Group.joins(:projects).where(projects: { id: project_id })
@@ -137,7 +148,7 @@ module Ci
       from_union(
         [
           belonging_to_project(project_id),
-          project.group_runners_enabled? ? belonging_to_parent_group_of_project(project_id) : nil,
+          project.group_runners_enabled? ? belonging_to_parent_groups_of_project(project_id) : nil,
           project.shared_runners
         ].compact,
         remove_duplicates: false
@@ -185,6 +196,7 @@ module Ci
     scope :order_token_expires_at_asc, -> { order(token_expires_at: :asc) }
     scope :order_token_expires_at_desc, -> { order(token_expires_at: :desc) }
     scope :with_tags, -> { preload(:tags) }
+    scope :with_creator, -> { preload(:creator) }
 
     validate :tag_constraints
     validates :access_level, presence: true
@@ -203,16 +215,14 @@ module Ci
     cached_attr_reader :version, :revision, :platform, :architecture, :ip_address, :contacted_at, :executor_type
 
     chronic_duration_attr :maximum_timeout_human_readable, :maximum_timeout,
-        error_message: 'Maximum job timeout has a value which could not be accepted'
+      error_message: 'Maximum job timeout has a value which could not be accepted'
 
     validates :maximum_timeout, allow_nil: true,
-                                numericality: { greater_than_or_equal_to: 600,
-                                                message: 'needs to be at least 10 minutes' }
+      numericality: { greater_than_or_equal_to: 600, message: 'needs to be at least 10 minutes' }
 
     validates :public_projects_minutes_cost_factor, :private_projects_minutes_cost_factor,
       allow_nil: false,
-      numericality: { greater_than_or_equal_to: 0.0,
-                      message: 'needs to be non-negative' }
+      numericality: { greater_than_or_equal_to: 0.0, message: 'needs to be non-negative' }
 
     validates :config, json_schema: { filename: 'ci_runner_config' }
 
@@ -332,15 +342,10 @@ module Ci
     def stale?
       return false unless created_at
 
-      [created_at, contacted_at].compact.max < self.class.stale_deadline
+      [created_at, contacted_at].compact.max <= self.class.stale_deadline
     end
 
-    def status(legacy_mode = nil)
-      # TODO Deprecate legacy_mode in %16.0 and make it a no-op
-      #   (see https://gitlab.com/gitlab-org/gitlab/-/issues/360545)
-      # TODO Remove legacy_mode in %17.0
-      return deprecated_rest_status if legacy_mode == '14.5'
-
+    def status
       return :stale if stale?
       return :never_contacted unless contacted_at
 
@@ -434,7 +439,7 @@ module Ci
       ensure_runner_queue_value == value if value.present?
     end
 
-    def heartbeat(values)
+    def heartbeat(values, update_contacted_at: true)
       ##
       # We can safely ignore writes performed by a runner heartbeat. We do
       # not want to upgrade database connection proxy to use the primary
@@ -442,20 +447,18 @@ module Ci
       #
       ::Gitlab::Database::LoadBalancing::Session.without_sticky_writes do
         values = values&.slice(:version, :revision, :platform, :architecture, :ip_address, :config, :executor) || {}
-        values[:contacted_at] = Time.current
+        values[:contacted_at] = Time.current if update_contacted_at
         if values.include?(:executor)
           values[:executor_type] = EXECUTOR_NAME_TO_TYPES.fetch(values.delete(:executor), :unknown)
         end
 
-        cache_attributes(values)
+        new_version = values[:version]
+        schedule_runner_version_update(new_version) if new_version && values[:version] != version
 
-        # We save data without validation, it will always change due to `contacted_at`
-        if persist_cached_data?
-          version_updated = values.include?(:version) && values[:version] != version
+        merge_cache_attributes(values)
 
-          update_columns(values)
-          schedule_runner_version_update if version_updated
-        end
+        # We save data without validation, it will always change due to `contacted_at`
+        update_columns(values) if persist_cached_data?
       end
     end
 
@@ -488,15 +491,18 @@ module Ci
       end
     end
 
-    override :format_token
-    def format_token(token)
-      return token if registration_token_registration_type?
+    def ensure_manager(system_xid, &blk)
+      RunnerManager.safe_find_or_create_by!(runner_id: id, system_xid: system_xid.to_s, &blk) # rubocop: disable Performance/ActiveRecordSubtransactionMethods
+    end
 
-      "#{CREATED_RUNNER_TOKEN_PREFIX}#{token}"
+    def registration_available?
+      authenticated_user_registration_type? &&
+        created_at > REGISTRATION_AVAILABILITY_TIME.ago &&
+        !runner_managers.any?
     end
 
-    def ensure_machine(system_xid, &blk)
-      RunnerMachine.safe_find_or_create_by!(runner_id: id, system_xid: system_xid.to_s, &blk) # rubocop: disable Performance/ActiveRecordSubtransactionMethods
+    def gitlab_hosted?
+      Gitlab.com? && instance_type?
     end
 
     private
@@ -586,7 +592,7 @@ module Ci
     end
 
     def exactly_one_group
-      unless runner_namespaces.one?
+      unless runner_namespaces.size == 1
         errors.add(:runner, 'needs to be assigned to exactly one group')
       end
     end
@@ -594,10 +600,16 @@ module Ci
     # TODO Remove in 16.0 when runners are known to send a system_id
     # For now, heartbeats with version updates might result in two Sidekiq jobs being queued if a runner has a system_id
     # This is not a problem since the jobs are deduplicated on the version
-    def schedule_runner_version_update
-      return unless version
+    def schedule_runner_version_update(new_version)
+      return unless new_version && Gitlab::Ci::RunnerReleases.instance.enabled?
+
+      Ci::Runners::ProcessRunnerVersionUpdateWorker.perform_async(new_version)
+    end
+
+    def prefix_for_new_and_legacy_runner
+      return if registration_token_registration_type?
 
-      Ci::Runners::ProcessRunnerVersionUpdateWorker.perform_async(version)
+      CREATED_RUNNER_TOKEN_PREFIX
     end
   end
 end
diff --git a/app/models/ci/runner_machine.rb b/app/models/ci/runner_manager.rb
index e52659a011f..e36024d9f5b 100644
--- a/app/models/ci/runner_machine.rb
+++ b/app/models/ci/runner_manager.rb
@@ -1,23 +1,23 @@
 # frozen_string_literal: true
 
 module Ci
-  class RunnerMachine < Ci::ApplicationRecord
+  class RunnerManager < Ci::ApplicationRecord
     include FromUnion
     include RedisCacheable
     include Ci::HasRunnerExecutor
-    include IgnorableColumns
 
-    ignore_column :machine_xid, remove_with: '15.11', remove_after: '2022-03-22'
+    # For legacy reasons, the table name is ci_runner_machines in the database
+    self.table_name = 'ci_runner_machines'
 
     # The `UPDATE_CONTACT_COLUMN_EVERY` defines how often the Runner Machine DB entry can be updated
-    UPDATE_CONTACT_COLUMN_EVERY = 40.minutes..55.minutes
+    UPDATE_CONTACT_COLUMN_EVERY = (40.minutes)..(55.minutes)
 
     belongs_to :runner
 
-    has_many :build_metadata, class_name: 'Ci::BuildMetadata'
-    has_many :builds, through: :build_metadata, class_name: 'Ci::Build'
-    belongs_to :runner_version, inverse_of: :runner_machines, primary_key: :version, foreign_key: :version,
-               class_name: 'Ci::RunnerVersion'
+    has_many :runner_manager_builds, inverse_of: :runner_manager, class_name: 'Ci::RunnerManagerBuild'
+    has_many :builds, through: :runner_manager_builds, class_name: 'Ci::Build'
+    belongs_to :runner_version, inverse_of: :runner_managers, primary_key: :version, foreign_key: :version,
+      class_name: 'Ci::RunnerVersion'
 
     validates :runner, presence: true
     validates :system_xid, presence: true, length: { maximum: 64 }
@@ -30,7 +30,7 @@ module Ci
 
     cached_attr_reader :version, :revision, :platform, :architecture, :ip_address, :contacted_at, :executor_type
 
-    # The `STALE_TIMEOUT` constant defines the how far past the last contact or creation date a runner machine
+    # The `STALE_TIMEOUT` constant defines the how far past the last contact or creation date a runner manager
     # will be considered stale
     STALE_TIMEOUT = 7.days
 
@@ -44,7 +44,15 @@ module Ci
         remove_duplicates: false).where(created_some_time_ago)
     end
 
-    def heartbeat(values)
+    def self.online_contact_time_deadline
+      Ci::Runner.online_contact_time_deadline
+    end
+
+    def self.stale_deadline
+      STALE_TIMEOUT.ago
+    end
+
+    def heartbeat(values, update_contacted_at: true)
       ##
       # We can safely ignore writes performed by a runner heartbeat. We do
       # not want to upgrade database connection proxy to use the primary
@@ -52,24 +60,40 @@ module Ci
       #
       ::Gitlab::Database::LoadBalancing::Session.without_sticky_writes do
         values = values&.slice(:version, :revision, :platform, :architecture, :ip_address, :config, :executor) || {}
-        values[:contacted_at] = Time.current
+        values[:contacted_at] = Time.current if update_contacted_at
         if values.include?(:executor)
           values[:executor_type] = Ci::Runner::EXECUTOR_NAME_TO_TYPES.fetch(values.delete(:executor), :unknown)
         end
 
-        version_changed = values.include?(:version) && values[:version] != version
+        new_version = values[:version]
+        schedule_runner_version_update(new_version) if new_version && values[:version] != version
 
-        cache_attributes(values)
-
-        schedule_runner_version_update if version_changed
+        merge_cache_attributes(values)
 
         # We save data without validation, it will always change due to `contacted_at`
         update_columns(values) if persist_cached_data?
       end
     end
 
+    def status
+      return :stale if stale?
+      return :never_contacted unless contacted_at
+
+      online? ? :online : :offline
+    end
+
     private
 
+    def online?
+      contacted_at && contacted_at > self.class.online_contact_time_deadline
+    end
+
+    def stale?
+      return false unless created_at
+
+      [created_at, contacted_at].compact.max <= self.class.stale_deadline
+    end
+
     def persist_cached_data?
       # Use a random threshold to prevent beating DB updates.
       contacted_at_max_age = Random.rand(UPDATE_CONTACT_COLUMN_EVERY)
@@ -79,10 +103,10 @@ module Ci
         (Time.current - real_contacted_at) >= contacted_at_max_age
     end
 
-    def schedule_runner_version_update
-      return unless version
+    def schedule_runner_version_update(new_version)
+      return unless new_version && Gitlab::Ci::RunnerReleases.instance.enabled?
 
-      Ci::Runners::ProcessRunnerVersionUpdateWorker.perform_async(version)
+      Ci::Runners::ProcessRunnerVersionUpdateWorker.perform_async(new_version)
     end
   end
 end
diff --git a/app/models/ci/runner_manager_build.rb b/app/models/ci/runner_manager_build.rb
new file mode 100644
index 00000000000..322c5ae3a68
--- /dev/null
+++ b/app/models/ci/runner_manager_build.rb
@@ -0,0 +1,29 @@
+# frozen_string_literal: true
+
+module Ci
+  class RunnerManagerBuild < Ci::ApplicationRecord
+    include Ci::Partitionable
+
+    self.table_name = :p_ci_runner_machine_builds
+    self.primary_key = :build_id
+
+    partitionable scope: :build, partitioned: true
+
+    alias_attribute :runner_manager_id, :runner_machine_id
+
+    belongs_to :build, inverse_of: :runner_manager_build, class_name: 'Ci::Build'
+    belongs_to :runner_manager, foreign_key: :runner_machine_id, inverse_of: :runner_manager_builds,
+      class_name: 'Ci::RunnerManager'
+
+    validates :build, presence: true
+    validates :runner_manager, presence: true
+
+    scope :for_build, ->(build_id) { where(build_id: build_id) }
+
+    def self.pluck_build_id_and_runner_manager_id
+      select(:build_id, :runner_manager_id)
+        .pluck(:build_id, :runner_manager_id)
+        .to_h
+    end
+  end
+end
diff --git a/app/models/ci/runner_version.rb b/app/models/ci/runner_version.rb
index ec42f46b165..03b50f13989 100644
--- a/app/models/ci/runner_version.rb
+++ b/app/models/ci/runner_version.rb
@@ -3,9 +3,8 @@
 module Ci
   class RunnerVersion < Ci::ApplicationRecord
     include EachBatch
-    include EnumWithNil
 
-    enum_with_nil status: {
+    enum status: {
       not_processed: nil,
       invalid_version: -1,
       unavailable: 1,
@@ -20,7 +19,7 @@ module Ci
       recommended: 'Upgrade is available and recommended for the runner.'
     }.freeze
 
-    has_many :runner_machines, inverse_of: :runner_version, foreign_key: :version, class_name: 'Ci::RunnerMachine'
+    has_many :runner_managers, inverse_of: :runner_version, foreign_key: :version, class_name: 'Ci::RunnerManager'
 
     # This scope returns all versions that might need recalculating. For instance, once a version is considered
     # :recommended, it normally doesn't change status even if the instance is upgraded
diff --git a/app/models/ci/running_build.rb b/app/models/ci/running_build.rb
index 43214b0c336..e6f80658f5d 100644
--- a/app/models/ci/running_build.rb
+++ b/app/models/ci/running_build.rb
@@ -24,10 +24,12 @@ module Ci
         raise ArgumentError, 'build has not been picked by a shared runner'
       end
 
-      entry = self.new(build: build,
-                       project: build.project,
-                       runner: build.runner,
-                       runner_type: build.runner.runner_type)
+      entry = self.new(
+        build: build,
+        project: build.project,
+        runner: build.runner,
+        runner_type: build.runner.runner_type
+      )
 
       entry.validate!
 
diff --git a/app/models/ci/sources/pipeline.rb b/app/models/ci/sources/pipeline.rb
index 855e68d1db1..719d19f4169 100644
--- a/app/models/ci/sources/pipeline.rb
+++ b/app/models/ci/sources/pipeline.rb
@@ -10,6 +10,7 @@ module Ci
 
       belongs_to :project, class_name: "::Project"
       belongs_to :pipeline, class_name: "Ci::Pipeline", inverse_of: :source_pipeline
+      belongs_to :build, class_name: "Ci::Build", foreign_key: :source_job_id, inverse_of: :sourced_pipelines
 
       belongs_to :source_project, class_name: "::Project", foreign_key: :source_project_id
       belongs_to :source_job, class_name: "CommitStatus", foreign_key: :source_job_id
diff --git a/app/models/ci/stage.rb b/app/models/ci/stage.rb
index 46a9e3f6494..d61760bd0fc 100644
--- a/app/models/ci/stage.rb
+++ b/app/models/ci/stage.rb
@@ -27,6 +27,7 @@ module Ci
     has_many :processables, class_name: 'Ci::Processable', foreign_key: :stage_id, inverse_of: :ci_stage
     has_many :builds, foreign_key: :stage_id, inverse_of: :ci_stage
     has_many :bridges, foreign_key: :stage_id, inverse_of: :ci_stage
+    has_many :generic_commit_statuses, foreign_key: :stage_id, inverse_of: :ci_stage
 
     scope :ordered, -> { order(position: :asc) }
     scope :in_pipelines, ->(pipelines) { where(pipeline: pipelines) }
@@ -111,12 +112,12 @@ module Ci
         when 'scheduled' then delay
         when 'skipped', nil then skip
         else
-          raise Ci::HasStatus::UnknownStatusError,
-                "Unknown status `#{new_status}`"
+          raise Ci::HasStatus::UnknownStatusError, "Unknown status `#{new_status}`"
         end
       end
     end
 
+    # This will be removed with ci_remove_ensure_stage_service
     def update_legacy_status
       set_status(latest_stage_status.to_s)
     end
@@ -150,6 +151,7 @@ module Ci
       blocked? || skipped?
     end
 
+    # This will be removed with ci_remove_ensure_stage_service
     def latest_stage_status
       statuses.latest.composite_status || 'skipped'
     end
diff --git a/app/models/ci/trigger.rb b/app/models/ci/trigger.rb
index 1b2a7dc3fe4..58da1b4bd7e 100644
--- a/app/models/ci/trigger.rb
+++ b/app/models/ci/trigger.rb
@@ -8,7 +8,7 @@ module Ci
 
     TRIGGER_TOKEN_PREFIX = 'glptt-'
 
-    ignore_column :ref, remove_with: '15.4', remove_after: '2022-08-22'
+    ignore_column :ref, remove_with: '16.1', remove_after: '2023-05-22'
 
     self.limit_name = 'pipeline_triggers'
     self.limit_scope = :project
@@ -26,8 +26,7 @@ module Ci
       mode: :per_attribute_iv,
       algorithm: 'aes-256-gcm',
       key: Settings.attr_encrypted_db_key_base_32,
-      encode: false,
-      encode_vi: false
+      encode: false
 
     before_validation :set_default_values