Rearchitect Fixed Email Notification (V2)

This commit refactors it

app/helpers/notifications_helper.rb

@@ -122,6 +122,6 @@ module NotificationsHelper
   end
 
   def notification_event_disabled?(event)
-    event == :fixed_pipeline && Feature.disabled?(:ci_pipeline_fixed_notifications)
+    event == :fixed_pipeline && !Gitlab::Ci::Features.pipeline_fixed_notifications?
   end
 end

app/models/ci/pipeline.rb

@@ -31,6 +31,7 @@ module Ci
     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: ->(s) do
       s&.project&.all_pipelines&.maximum(:iid) || s&.project&.all_pipelines&.count
@@ -66,13 +67,6 @@ module Ci
 
     has_one :source_pipeline, class_name: 'Ci::Sources::Pipeline', inverse_of: :pipeline
 
-    has_one :ref_status, ->(pipeline) {
-      # We use .read_attribute to save 1 extra unneeded query to load the :project.
-      unscope(:where)
-        .where(project_id: pipeline.read_attribute(:project_id), ref: pipeline.ref, tag: pipeline.tag)
-      # Sadly :inverse_of is not supported (yet) by Rails for composite PKs.
-    }, class_name: 'Ci::Ref', inverse_of: :pipelines
-
     has_one :chat_data, class_name: 'Ci::PipelineChatData'
 
     has_many :triggered_pipelines, through: :sourced_pipelines, source: :pipeline
@@ -237,12 +231,10 @@ module Ci
       end
 
       after_transition any => [:success, :failed] do |pipeline|
+        ref_status = pipeline.ci_ref&.update_status_by!(pipeline)
+
         pipeline.run_after_commit do
-          if Feature.enabled?(:ci_pipeline_fixed_notifications)
-            PipelineUpdateCiRefStatusWorker.perform_async(pipeline.id)
-          else
-            PipelineNotificationWorker.perform_async(pipeline.id)
-          end
+          PipelineNotificationWorker.perform_async(pipeline.id, ref_status: ref_status)
         end
       end
 
@@ -975,6 +967,12 @@ module Ci
       processables.populate_scheduling_type!
     end
 
+    def ensure_ci_ref!
+      return unless Gitlab::Ci::Features.pipeline_fixed_notifications?
+
+      self.ci_ref = Ci::Ref.ensure_for(self)
+    end
+
     private
 
     def pipeline_data

app/models/ci/ref.rb

@@ -3,21 +3,62 @@
 module Ci
   class Ref < ApplicationRecord
     extend Gitlab::Ci::Model
+    include Gitlab::OptimisticLocking
 
-    STATUSES = %w[success failed fixed].freeze
-
-    belongs_to :project
-    belongs_to :last_updated_by_pipeline, foreign_key: :last_updated_by_pipeline_id, class_name: 'Ci::Pipeline'
-    # ActiveRecord doesn't support composite FKs for this reason we have to do the 'unscope(:where)'
-    # hack.
-    has_many :pipelines, ->(ref) {
-      # We use .read_attribute to save 1 extra unneeded query to load the :project.
-      unscope(:where)
-        .where(ref: ref.ref, project_id: ref.read_attribute(:project_id), tag: ref.tag)
-      # Sadly :inverse_of is not supported (yet) by Rails for composite PKs.
-    }, inverse_of: :ref_status
-
-    validates :status, inclusion: { in: STATUSES }
-    validates :last_updated_by_pipeline, presence: true
+    FAILING_STATUSES = %w[failed broken still_failing].freeze
+
+    belongs_to :project, inverse_of: :ci_refs
+    has_many :pipelines, class_name: 'Ci::Pipeline', foreign_key: :ci_ref_id, inverse_of: :ci_ref
+
+    state_machine :status, initial: :unknown do
+      event :succeed do
+        transition unknown: :success
+        transition fixed: :success
+        transition %i[failed broken still_failing] => :fixed
+      end
+
+      event :do_fail do
+        transition unknown: :failed
+        transition %i[failed broken] => :still_failing
+        transition %i[success fixed] => :broken
+      end
+
+      state :unknown, value: 0
+      state :success, value: 1
+      state :failed, value: 2
+      state :fixed, value: 3
+      state :broken, value: 4
+      state :still_failing, value: 5
+    end
+
+    class << self
+      def ensure_for(pipeline)
+        safe_find_or_create_by(project_id: pipeline.project_id,
+                               ref_path: pipeline.source_ref_path)
+      end
+
+      def failing_state?(status_name)
+        FAILING_STATUSES.include?(status_name)
+      end
+    end
+
+    def last_finished_pipeline_id
+      Ci::Pipeline.where(ci_ref_id: self.id).finished.order(id: :desc).select(:id).take&.id
+    end
+
+    def update_status_by!(pipeline)
+      return unless Gitlab::Ci::Features.pipeline_fixed_notifications?
+
+      retry_lock(self) do
+        next unless last_finished_pipeline_id == pipeline.id
+
+        case pipeline.status
+        when 'success' then self.succeed
+        when 'failed' then self.do_fail
+        end
+
+        self.status_name
+      end
+    end
   end
 end

app/models/project.rb

@@ -282,7 +282,7 @@ class Project < ApplicationRecord
           class_name: 'Ci::Pipeline',
           inverse_of: :project
   has_many :stages, class_name: 'Ci::Stage', inverse_of: :project
-  has_many :ci_refs, class_name: 'Ci::Ref'
+  has_many :ci_refs, class_name: 'Ci::Ref', inverse_of: :project
 
   # Ci::Build objects store data on the file system such as artifact files and
   # build traces. Currently there's no efficient way of removing this data in

app/services/ci/update_ci_ref_status_service.rb

 # frozen_string_literal: true
 
+# NOTE: This class is unused and to be removed in 13.1~
 module Ci
   class UpdateCiRefStatusService
     include Gitlab::OptimisticLocking

app/services/notification_service.rb

@@ -447,14 +447,14 @@ class NotificationService
     # from the PipelinesEmailService integration.
     return if pipeline.project.emails_disabled?
 
-    ref_status ||= pipeline.status
-    email_template = "pipeline_#{ref_status}_email"
+    status = pipeline_notification_status(ref_status, pipeline)
+    email_template = "pipeline_#{status}_email"
 
     return unless mailer.respond_to?(email_template)
 
     recipients ||= notifiable_users(
       [pipeline.user], :watch,
-      custom_action: :"#{ref_status}_pipeline",
+      custom_action: :"#{status}_pipeline",
       target: pipeline
     ).map do |user|
       user.notification_email_for(pipeline.project.group)
@@ -661,6 +661,16 @@ class NotificationService
 
   private
 
+  def pipeline_notification_status(ref_status, pipeline)
+    if Ci::Ref.failing_state?(ref_status)
+      'failed'
+    elsif ref_status
+      ref_status
+    else
+      pipeline.status
+    end
+  end
+
   def owners_and_maintainers_without_invites(project)
     recipients = project.members.active_without_invites_and_requests.owners_and_maintainers
 

app/workers/pipeline_notification_worker.rb

@@ -7,10 +7,10 @@ class PipelineNotificationWorker # rubocop:disable Scalability/IdempotentWorker
   urgency :high
   worker_resource_boundary :cpu
 
-  # rubocop: disable CodeReuse/ActiveRecord
   def perform(pipeline_id, args = {})
     case args
     when Hash
+      args = args.with_indifferent_access
       ref_status = args[:ref_status]
       recipients = args[:recipients]
     else # TODO: backward compatible interface, can be removed in 12.10
@@ -18,10 +18,9 @@ class PipelineNotificationWorker # rubocop:disable Scalability/IdempotentWorker
       ref_status = nil
     end
 
-    pipeline = Ci::Pipeline.find_by(id: pipeline_id)
+    pipeline = Ci::Pipeline.find_by_id(pipeline_id)
     return unless pipeline
 
     NotificationService.new.pipeline_finished(pipeline, ref_status: ref_status, recipients: recipients)
   end
-  # rubocop: enable CodeReuse/ActiveRecord
 end

app/workers/pipeline_update_ci_ref_status_worker.rb

 # frozen_string_literal: true
 
+# NOTE: This class is unused and to be removed in 13.1~
 class PipelineUpdateCiRefStatusWorker # rubocop:disable Scalability/IdempotentWorker
   include ApplicationWorker
   include PipelineQueue

changelogs/unreleased/rearchitect-fixed-pipelines-notification-v2.yml

+---
+title: Make Fixed Email Notification Generally Available
+merge_request: 28338
+author: jacopo-beschi
+type: added

db/migrate/20200330203826_drop_fk_in_ci_ref.rb

+# frozen_string_literal: true
+
+class DropFkInCiRef < ActiveRecord::Migration[6.0]
+  include Gitlab::Database::MigrationHelpers
+
+  DOWNTIME = false
+
+  disable_ddl_transaction!
+
+  def up
+    with_lock_retries do
+      remove_foreign_key_if_exists :ci_refs, column: :project_id
+    end
+    with_lock_retries do
+      remove_foreign_key_if_exists :ci_refs, column: :last_updated_by_pipeline_id
+    end
+  end
+
+  def down
+    add_foreign_key_if_not_exists :ci_refs, :projects, column: :project_id, on_delete: :cascade
+    add_foreign_key_if_not_exists :ci_refs, :ci_pipelines, column: :last_updated_by_pipeline_id, on_delete: :nullify
+  end
+
+  private
+
+  def add_foreign_key_if_not_exists(source, target, column:, on_delete:)
+    return unless table_exists?(source)
+    return if foreign_key_exists?(source, target, column: column)
+
+    add_concurrent_foreign_key(source, target, column: column, on_delete: on_delete)
+  end
+end

db/migrate/20200330203837_recreate_ci_ref.rb

+# frozen_string_literal: true
+
+class RecreateCiRef < ActiveRecord::Migration[6.0]
+  include Gitlab::Database::MigrationHelpers
+
+  DOWNTIME = false
+
+  UNKNOWN_STATUS = 0
+
+  def up
+    with_lock_retries do
+      drop_table :ci_refs
+
+      create_table :ci_refs do |t|
+        t.references :project, null: false, index: false, foreign_key: { on_delete: :cascade }, type: :bigint
+        t.integer :lock_version, null: false, default: 0
+        t.integer :status, null: false, limit: 2, default: UNKNOWN_STATUS
+        t.text :ref_path, null: false # rubocop: disable Migration/AddLimitToTextColumns
+        t.index [:project_id, :ref_path], unique: true
+      end
+    end
+  end
+
+  def down
+    with_lock_retries do
+      drop_table :ci_refs
+
+      create_table :ci_refs do |t|
+        t.references :project, null: false, index: false, foreign_key: { on_delete: :cascade }, type: :integer
+        t.integer :lock_version, default: 0
+        t.integer :last_updated_by_pipeline_id
+        t.boolean :tag, default: false, null: false
+        t.string :ref, null: false, limit: 255
+        t.string :status, null: false, limit: 255
+        t.foreign_key :ci_pipelines, column: :last_updated_by_pipeline_id, on_delete: :nullify
+        t.index [:project_id, :ref, :tag], unique: true
+        t.index [:last_updated_by_pipeline_id]
+      end
+    end
+  end
+end

db/migrate/20200331103637_add_ci_ref_id_to_ci_pipelines.rb

+# frozen_string_literal: true
+
+class AddCiRefIdToCiPipelines < ActiveRecord::Migration[6.0]
+  include Gitlab::Database::MigrationHelpers
+
+  DOWNTIME = false
+
+  def up
+    with_lock_retries do
+      add_column :ci_pipelines, :ci_ref_id, :bigint
+    end
+  end
+
+  def down
+    with_lock_retries do
+      remove_column :ci_pipelines, :ci_ref_id, :bigint
+    end
+  end
+end

db/migrate/20200331113728_add_index_to_ci_ref_id.rb

+# frozen_string_literal: true
+
+class AddIndexToCiRefId < ActiveRecord::Migration[6.0]
+  include Gitlab::Database::MigrationHelpers
+
+  DOWNTIME = false
+  disable_ddl_transaction!
+
+  def up
+    add_concurrent_index :ci_pipelines, [:ci_ref_id], where: 'ci_ref_id IS NOT NULL'
+  end
+
+  def down
+    remove_concurrent_index :ci_pipelines, [:ci_ref_id], where: 'ci_ref_id IS NOT NULL'
+  end
+end

db/migrate/20200331113738_add_fk_to_ci_ref_id.rb

+# frozen_string_literal: true
+
+class AddFkToCiRefId < ActiveRecord::Migration[6.0]
+  include Gitlab::Database::MigrationHelpers
+
+  DOWNTIME = false
+  disable_ddl_transaction!
+
+  def up
+    add_concurrent_foreign_key :ci_pipelines, :ci_refs, column: :ci_ref_id, on_delete: :nullify
+  end
+
+  def down
+    with_lock_retries do
+      remove_foreign_key_if_exists :ci_pipelines, column: :ci_ref_id
+    end
+  end
+end

db/structure.sql

@@ -1322,7 +1322,8 @@ CREATE TABLE public.ci_pipelines (
     merge_request_id integer,
     source_sha bytea,
     target_sha bytea,
-    external_pull_request_id bigint
+    external_pull_request_id bigint,
+    ci_ref_id bigint
 );
 
 CREATE TABLE public.ci_pipelines_config (
@@ -1350,12 +1351,10 @@ ALTER SEQUENCE public.ci_pipelines_id_seq OWNED BY public.ci_pipelines.id;
 
 CREATE TABLE public.ci_refs (
     id bigint NOT NULL,
-    project_id integer NOT NULL,
-    lock_version integer DEFAULT 0,
-    last_updated_by_pipeline_id integer,
-    tag boolean DEFAULT false NOT NULL,
-    ref character varying(255) NOT NULL,
-    status character varying(255) NOT NULL
+    project_id bigint NOT NULL,
+    lock_version integer DEFAULT 0 NOT NULL,
+    status smallint DEFAULT 0 NOT NULL,
+    ref_path text NOT NULL
 );
 
 CREATE SEQUENCE public.ci_refs_id_seq
@@ -9424,6 +9423,8 @@ CREATE INDEX index_ci_pipelines_config_on_pipeline_id ON public.ci_pipelines_con
 
 CREATE INDEX index_ci_pipelines_on_auto_canceled_by_id ON public.ci_pipelines USING btree (auto_canceled_by_id);
 
+CREATE INDEX index_ci_pipelines_on_ci_ref_id ON public.ci_pipelines USING btree (ci_ref_id) WHERE (ci_ref_id IS NOT NULL);
+
 CREATE INDEX index_ci_pipelines_on_external_pull_request_id ON public.ci_pipelines USING btree (external_pull_request_id) WHERE (external_pull_request_id IS NOT NULL);
 
 CREATE INDEX index_ci_pipelines_on_merge_request_id ON public.ci_pipelines USING btree (merge_request_id) WHERE (merge_request_id IS NOT NULL);
@@ -9452,9 +9453,7 @@ CREATE INDEX index_ci_pipelines_on_status ON public.ci_pipelines USING btree (st
 
 CREATE INDEX index_ci_pipelines_on_user_id_and_created_at ON public.ci_pipelines USING btree (user_id, created_at);
 
-CREATE INDEX index_ci_refs_on_last_updated_by_pipeline_id ON public.ci_refs USING btree (last_updated_by_pipeline_id);
-
-CREATE UNIQUE INDEX index_ci_refs_on_project_id_and_ref_and_tag ON public.ci_refs USING btree (project_id, ref, tag);
+CREATE UNIQUE INDEX index_ci_refs_on_project_id_and_ref_path ON public.ci_refs USING btree (project_id, ref_path);
 
 CREATE UNIQUE INDEX index_ci_resource_groups_on_project_id_and_key ON public.ci_resource_groups USING btree (project_id, key);
 
@@ -11610,6 +11609,9 @@ ALTER TABLE ONLY public.lists
 ALTER TABLE ONLY public.metrics_users_starred_dashboards
     ADD CONSTRAINT fk_d76a2b9a8c FOREIGN KEY (project_id) REFERENCES public.projects(id) ON DELETE CASCADE;
 
+ALTER TABLE ONLY public.ci_pipelines
+    ADD CONSTRAINT fk_d80e161c54 FOREIGN KEY (ci_ref_id) REFERENCES public.ci_refs(id) ON DELETE SET NULL;
+
 ALTER TABLE ONLY public.system_note_metadata
     ADD CONSTRAINT fk_d83a918cb1 FOREIGN KEY (note_id) REFERENCES public.notes(id) ON DELETE CASCADE;
 
@@ -11859,9 +11861,6 @@ ALTER TABLE ONLY public.epic_user_mentions
 ALTER TABLE ONLY public.approver_groups
     ADD CONSTRAINT fk_rails_1cdcbd7723 FOREIGN KEY (group_id) REFERENCES public.namespaces(id) ON DELETE CASCADE;
 
-ALTER TABLE ONLY public.ci_refs
-    ADD CONSTRAINT fk_rails_1da48d19ce FOREIGN KEY (last_updated_by_pipeline_id) REFERENCES public.ci_pipelines(id) ON DELETE SET NULL;
-
 ALTER TABLE ONLY public.packages_tags
     ADD CONSTRAINT fk_rails_1dfc868911 FOREIGN KEY (package_id) REFERENCES public.packages_packages(id) ON DELETE CASCADE;
 
@@ -13634,6 +13633,11 @@ COPY "schema_migrations" (version) FROM STDIN;
 20200330121000
 20200330123739
 20200330132913
+20200330203826
+20200330203837
+20200331103637
+20200331113728
+20200331113738
 20200331132103
 20200331195952
 20200331220930

ee/spec/models/ci/minutes/context_spec.rb

@@ -4,7 +4,7 @@ require 'spec_helper'
 
 RSpec.describe Ci::Minutes::Context do
   let_it_be(:group) { create(:group) }
-  let(:project) { build(:project, namespace: group) }
+  let_it_be(:project) { create(:project, namespace: group) }
 
   describe 'delegation' do
     subject { described_class.new(project, group) }

ee/spec/services/ci/compare_container_scanning_reports_service_spec.rb

@@ -2,10 +2,10 @@
 
 require 'spec_helper'
 
-RSpec.describe Ci::CompareContainerScanningReportsService do
+describe Ci::CompareContainerScanningReportsService do
+  let_it_be(:project) { create(:project, :repository) }
   let(:current_user) { build(:user, :admin) }
   let(:service) { described_class.new(project, current_user) }
-  let(:project) { build(:project, :repository) }
 
   before do
     stub_licensed_features(container_scanning: true)

ee/spec/services/ci/compare_dast_reports_service_spec.rb

@@ -2,10 +2,10 @@
 
 require 'spec_helper'
 
-RSpec.describe Ci::CompareDastReportsService do
+describe Ci::CompareDastReportsService do
+  let_it_be(:project) { create(:project, :repository) }
   let(:current_user) { build(:user, :admin) }
   let(:service) { described_class.new(project, current_user) }
-  let(:project) { build(:project, :repository) }
 
   before do
     stub_licensed_features(container_scanning: true, dast: true)

ee/spec/services/ci/compare_dependency_scanning_reports_service_spec.rb

@@ -2,10 +2,10 @@
 
 require 'spec_helper'
 
-RSpec.describe Ci::CompareDependencyScanningReportsService do
+describe Ci::CompareDependencyScanningReportsService do
+  let_it_be(:project) { create(:project, :repository) }
   let(:current_user) { build(:user, :admin) }
   let(:service) { described_class.new(project, current_user) }
-  let(:project) { build(:project, :repository) }
 
   before do
     stub_licensed_features(dependency_scanning: true)

ee/spec/services/ci/compare_license_scanning_reports_service_spec.rb

@@ -2,9 +2,9 @@
 
 require 'spec_helper'
 
-RSpec.describe Ci::CompareLicenseScanningReportsService do
+describe Ci::CompareLicenseScanningReportsService do
+  let_it_be(:project) { create(:project, :repository) }
   let(:service) { described_class.new(project, nil) }
-  let(:project) { build(:project, :repository) }
 
   before do
     stub_licensed_features(license_scanning: true)

ee/spec/services/ci/compare_sast_reports_service_spec.rb

@@ -2,10 +2,10 @@
 
 require 'spec_helper'
 
-RSpec.describe Ci::CompareSastReportsService do
+describe Ci::CompareSastReportsService do
+  let_it_be(:project) { create(:project, :repository) }
   let(:current_user) { build(:user, :admin) }
   let(:service) { described_class.new(project, current_user) }
-  let(:project) { build(:project, :repository) }
 
   before do
     stub_licensed_features(container_scanning: true)

ee/spec/services/ci/compare_secret_detection_reports_service_spec.rb

@@ -5,7 +5,8 @@ require 'spec_helper'
 RSpec.describe Ci::CompareSecretDetectionReportsService do
   let(:current_user) { build(:user, :admin) }
   let(:service) { described_class.new(project, current_user) }
-  let(:project) { build(:project, :repository) }
+
+  let_it_be(:project) { create(:project, :repository) }
 
   before do
     stub_licensed_features(container_scanning: true)

ee/spec/services/ci/pipeline_bridge_status_service_spec.rb

@@ -2,9 +2,9 @@
 
 require 'spec_helper'
 
-RSpec.describe Ci::PipelineBridgeStatusService do
+describe Ci::PipelineBridgeStatusService do
+  let_it_be(:project) { create(:project) }
   let(:user) { build(:user) }
-  let(:project) { build(:project) }
   let(:pipeline) { build(:ci_pipeline, project: project) }
 
   describe '#execute' do

lib/gitlab/ci/features.rb

@@ -21,6 +21,10 @@ module Gitlab
       def self.instance_level_variables_limit_enabled?
         ::Feature.enabled?(:ci_instance_level_variables_limit, default_enabled: true)
       end
+
+      def self.pipeline_fixed_notifications?
+        ::Feature.enabled?(:ci_pipeline_fixed_notifications)
+      end
     end
   end
 end

lib/gitlab/ci/pipeline/chain/seed.rb

@@ -13,6 +13,7 @@ module Gitlab
 
             # Allocate next IID. This operation must be outside of transactions of pipeline creations.
             pipeline.ensure_project_iid!
+            pipeline.ensure_ci_ref!
 
             # Protect the pipeline. This is assigned in Populate instead of
             # Build to prevent erroring out on ambiguous refs.

lib/gitlab/import_export/project/import_export.yml

@@ -312,6 +312,7 @@ excluded_attributes:
     - :pipeline_schedule_id
     - :merge_request_id
     - :external_pull_request_id
+    - :ci_ref_id
   stages:
     - :pipeline_id
   merge_access_levels:

spec/factories/ci/pipelines.rb

@@ -21,6 +21,12 @@ FactoryBot.define do
     end
 
     factory :ci_pipeline do
+      transient { ci_ref_presence { true } }
+
+      after(:build) do |pipeline, evaluator|
+        pipeline.ensure_ci_ref! if evaluator.ci_ref_presence && pipeline.ci_ref_id.nil?
+      end
+
       trait :invalid do
         status { :failed }
         yaml_errors { 'invalid YAML' }

spec/factories/ci/ref.rb

@@ -2,15 +2,7 @@
 
 FactoryBot.define do
   factory :ci_ref, class: 'Ci::Ref' do
-    ref { 'master' }
-    status { :success }
-    tag { false }
+    ref_path { 'refs/heads/master' }
     project
-
-    before(:create) do |ref, evaluator|
-      next if ref.pipelines.exists?
-
-      ref.update!(last_updated_by_pipeline: create(:ci_pipeline, project: evaluator.project, ref: evaluator.ref, tag: evaluator.tag, status: evaluator.status))
-    end
   end
 end

spec/lib/gitlab/ci/pipeline/chain/seed_spec.rb

@@ -39,6 +39,10 @@ describe Gitlab::Ci::Pipeline::Chain::Seed do
       expect(pipeline.iid).to be_present
     end
 
+    it 'ensures ci_ref' do
+      expect(pipeline.ci_ref).to be_present
+    end
+
     it 'sets the seeds in the command object' do
       expect(command.stage_seeds).to all(be_a Gitlab::Ci::Pipeline::Seed::Base)
       expect(command.stage_seeds.count).to eq 1

spec/lib/gitlab/import_export/all_models.yml

@@ -179,6 +179,7 @@ merge_request_context_commits:
 ci_pipelines:
 - project
 - user
+- ci_ref
 - stages
 - statuses
 - latest_statuses_ordered_by_stage
@@ -221,6 +222,10 @@ ci_pipelines:
 - security_scans
 - daily_build_group_report_results
 - latest_builds
+- daily_report_results
+ci_refs:
+- project
+- ci_pipelines
 pipeline_variables:
 - pipeline
 stages:

spec/models/ci/pipeline_spec.rb

@@ -2641,38 +2641,34 @@ describe Ci::Pipeline, :mailer do
       end
     end
 
-    shared_examples 'enqueues the notification worker' do
-      it 'enqueues PipelineUpdateCiRefStatusWorker' do
-        expect(PipelineUpdateCiRefStatusWorker).to receive(:perform_async).with(pipeline.id)
-        expect(PipelineNotificationWorker).not_to receive(:perform_async).with(pipeline.id)
+    context 'with success pipeline' do
+      it_behaves_like 'sending a notification' do
+        before do
+          perform_enqueued_jobs do
+            pipeline.succeed
+          end
+        end
+      end
+
+      it 'enqueues PipelineNotificationWorker' do
+        expect(PipelineNotificationWorker)
+          .to receive(:perform_async).with(pipeline.id, ref_status: :success)
 
         pipeline.succeed
       end
 
-      context 'when ci_pipeline_fixed_notifications is disabled' do
+      context 'when pipeline is not the latest' do
         before do
-          stub_feature_flags(ci_pipeline_fixed_notifications: false)
+          create(:ci_pipeline, :success, project: project, ci_ref: pipeline.ci_ref)
         end
 
-        it 'enqueues PipelineNotificationWorker' do
-          expect(PipelineUpdateCiRefStatusWorker).not_to receive(:perform_async).with(pipeline.id)
-          expect(PipelineNotificationWorker).to receive(:perform_async).with(pipeline.id)
-
-          pipeline.succeed
-        end
-      end
-    end
+        it 'does not pass ref_status' do
+          expect(PipelineNotificationWorker)
+            .to receive(:perform_async).with(pipeline.id, ref_status: nil)
 
-    context 'with success pipeline' do
-      it_behaves_like 'sending a notification' do
-        before do
-          perform_enqueued_jobs do
-            pipeline.succeed
-          end
+          pipeline.succeed!
         end
       end
-
-      it_behaves_like 'enqueues the notification worker'
     end
 
     context 'with failed pipeline' do
@@ -2687,7 +2683,12 @@ describe Ci::Pipeline, :mailer do
         end
       end
 
-      it_behaves_like 'enqueues the notification worker'
+      it 'enqueues PipelineNotificationWorker' do
+        expect(PipelineNotificationWorker)
+          .to receive(:perform_async).with(pipeline.id, ref_status: :failed)
+
+        pipeline.drop
+      end
     end
 
     context 'with skipped pipeline' do
@@ -2711,6 +2712,69 @@ describe Ci::Pipeline, :mailer do
     end
   end
 
+  describe 'updates ci_ref when pipeline finished' do
+    context 'when ci_ref exists' do
+      let!(:pipeline) { create(:ci_pipeline, :running) }
+
+      it 'updates the ci_ref' do
+        expect(pipeline.ci_ref)
+          .to receive(:update_status_by!).with(pipeline).and_call_original
+
+        pipeline.succeed!
+      end
+    end
+
+    context 'when ci_ref does not exist' do
+      let!(:pipeline) { create(:ci_pipeline, :running, ci_ref_presence: false) }
+
+      it 'does not raise an exception' do
+        expect { pipeline.succeed! }.not_to raise_error
+      end
+    end
+  end
+
+  describe '#ensure_ci_ref!' do
+    subject { pipeline.ensure_ci_ref! }
+
+    shared_examples_for 'protected by feature flag' do
+      context 'when feature flag is disabled' do
+        before do
+          stub_feature_flags(ci_pipeline_fixed_notifications: false)
+        end
+
+        it 'does not do anything' do
+          expect(Ci::Ref).not_to receive(:ensure_for)
+
+          subject
+        end
+      end
+    end
+
+    context 'when ci_ref does not exist yet' do
+      let!(:pipeline) { create(:ci_pipeline, ci_ref_presence: false) }
+
+      it_behaves_like 'protected by feature flag'
+
+      it 'creates a new ci_ref and assigns it' do
+        expect { subject }.to change { Ci::Ref.count }.by(1)
+
+        expect(pipeline.ci_ref).to be_present
+      end
+    end
+
+    context 'when ci_ref already exists' do
+      let!(:pipeline) { create(:ci_pipeline) }
+
+      it_behaves_like 'protected by feature flag'
+
+      it 'fetches a new ci_ref and assigns it' do
+        expect { subject }.not_to change { Ci::Ref.count }
+
+        expect(pipeline.ci_ref).to be_present
+      end
+    end
+  end
+
   describe '#find_job_with_archive_artifacts' do
     let!(:old_job) { create(:ci_build, name: 'rspec', retried: true, pipeline: pipeline) }
     let!(:job_without_artifacts) { create(:ci_build, name: 'rspec', pipeline: pipeline) }

spec/models/ci/ref_spec.rb

@@ -4,8 +4,155 @@ require 'spec_helper'
 
 describe Ci::Ref do
   it { is_expected.to belong_to(:project) }
-  it { is_expected.to belong_to(:last_updated_by_pipeline) }
 
-  it { is_expected.to validate_inclusion_of(:status).in_array(%w[success failed fixed]) }
-  it { is_expected.to validate_presence_of(:last_updated_by_pipeline) }
+  describe '.ensure_for' do
+    let_it_be(:project) { create(:project, :repository) }
+
+    subject { described_class.ensure_for(pipeline) }
+
+    shared_examples_for 'ensures ci_ref' do
+      context 'when ci_ref already exists' do
+        let(:options) { {} }
+
+        it 'returns an existing ci_ref' do
+          expect { subject }.not_to change { described_class.count }
+
+          expect(subject).to eq(Ci::Ref.find_by(project_id: project.id, ref_path: expected_ref_path))
+        end
+      end
+
+      context 'when ci_ref does not exist yet' do
+        let(:options) { { ci_ref_presence: false } }
+
+        it 'creates a new ci_ref' do
+          expect { subject }.to change { described_class.count }.by(1)
+
+          expect(subject).to eq(Ci::Ref.find_by(project_id: project.id, ref_path: expected_ref_path))
+        end
+      end
+    end
+
+    context 'when pipeline is a branch pipeline' do
+      let!(:pipeline) { create(:ci_pipeline, ref: 'master', project: project, **options) }
+      let(:expected_ref_path) { 'refs/heads/master' }
+
+      it_behaves_like 'ensures ci_ref'
+    end
+
+    context 'when pipeline is a tag pipeline' do
+      let!(:pipeline) { create(:ci_pipeline, ref: 'v1.1.0', tag: true, project: project, **options) }
+      let(:expected_ref_path) { 'refs/tags/v1.1.0' }
+
+      it_behaves_like 'ensures ci_ref'
+    end
+
+    context 'when pipeline is a detached merge request pipeline' do
+      let(:merge_request) do
+        create(:merge_request, target_project: project, target_branch: 'master',
+                               source_project: project, source_branch: 'feature')
+      end
+
+      let!(:pipeline) do
+        create(:ci_pipeline, :detached_merge_request_pipeline, merge_request: merge_request, project: project, **options)
+      end
+
+      let(:expected_ref_path) { 'refs/heads/feature' }
+
+      it_behaves_like 'ensures ci_ref'
+    end
+  end
+
+  describe '#update_status_by!' do
+    subject { ci_ref.update_status_by!(pipeline) }
+
+    let!(:ci_ref) { create(:ci_ref) }
+
+    shared_examples_for 'no-op' do
+      it 'does nothing and returns nil' do
+        expect { subject }.not_to change { ci_ref.status_name }
+
+        is_expected.to be_nil
+      end
+    end
+
+    context 'when pipeline status is success or failed' do
+      using RSpec::Parameterized::TableSyntax
+
+      where(:pipeline_status, :current_ref_status, :expected_ref_status) do
+        :success   | :unknown       | :success
+        :success   | :success       | :success
+        :success   | :failed        | :fixed
+        :success   | :fixed         | :success
+        :success   | :broken        | :fixed
+        :success   | :still_failing | :fixed
+        :failed    | :unknown       | :failed
+        :failed    | :success       | :broken
+        :failed    | :failed        | :still_failing
+        :failed    | :fixed         | :broken
+        :failed    | :broken        | :still_failing
+        :failed    | :still_failing | :still_failing
+      end
+
+      with_them do
+        let(:ci_ref) { create(:ci_ref, status: described_class.state_machines[:status].states[current_ref_status].value) }
+        let(:pipeline) { create(:ci_pipeline, status: pipeline_status, ci_ref: ci_ref) }
+
+        it 'transitions the status via state machine' do
+          expect(subject).to eq(expected_ref_status)
+          expect(ci_ref.status_name).to eq(expected_ref_status)
+        end
+      end
+    end
+
+    context 'when pipeline status is success' do
+      let(:pipeline) { create(:ci_pipeline, :success, ci_ref: ci_ref) }
+
+      it 'updates the status' do
+        expect { subject }.to change { ci_ref.status_name }.from(:unknown).to(:success)
+
+        is_expected.to eq(:success)
+      end
+    end
+
+    context 'when pipeline status is canceled' do
+      let(:pipeline) { create(:ci_pipeline, status: :canceled, ci_ref: ci_ref) }
+
+      it { is_expected.to eq(:unknown) }
+    end
+
+    context 'when pipeline status is skipped' do
+      let(:pipeline) { create(:ci_pipeline, status: :skipped, ci_ref: ci_ref) }
+
+      it_behaves_like 'no-op'
+    end
+
+    context 'when pipeline status is not complete' do
+      let(:pipeline) { create(:ci_pipeline, :running, ci_ref: ci_ref) }
+
+      it_behaves_like 'no-op'
+    end
+
+    context 'when feature flag is disabled' do
+      let(:pipeline) { create(:ci_pipeline, :success, ci_ref: ci_ref) }
+
+      before do
+        stub_feature_flags(ci_pipeline_fixed_notifications: false)
+      end
+
+      it_behaves_like 'no-op'
+    end
+
+    context 'when pipeline is not the latest pipeline' do
+      let!(:pipeline) { create(:ci_pipeline, :success, ci_ref: ci_ref) }
+      let!(:latest_pipeline) { create(:ci_pipeline, :success, ci_ref: ci_ref) }
+
+      it_behaves_like 'no-op'
+    end
+
+    context 'when pipeline does not belong to the ci_ref' do
+      let(:pipeline) { create(:ci_pipeline, :success, ci_ref: create(:ci_ref)) }
+
+      it_behaves_like 'no-op'
+    end
+  end
 end

spec/services/ci/pipeline_bridge_status_service_spec.rb

@@ -4,7 +4,7 @@ require 'spec_helper'
 
 describe Ci::PipelineBridgeStatusService do
   let(:user) { build(:user) }
-  let(:project) { build(:project) }
+  let_it_be(:project) { create(:project) }
   let(:pipeline) { build(:ci_pipeline, project: project) }
 
   describe '#execute' do

spec/services/ci/update_ci_ref_status_service_spec.rb

-# frozen_string_literal: true
-
-require 'spec_helper'
-
-describe Ci::UpdateCiRefStatusService do
-  describe '#call' do
-    subject { described_class.new(pipeline) }
-
-    shared_examples 'creates ci_ref' do
-      it 'creates a ci_ref with the pipeline attributes' do
-        expect do
-          expect(subject.call).to eq(true)
-        end.to change { Ci::Ref.count }.by(1)
-
-        created_ref = pipeline.reload.ref_status
-        %w[ref tag project status].each do |attr|
-          expect(created_ref[attr]).to eq(pipeline[attr])
-        end
-      end
-
-      it 'calls PipelineNotificationWorker pasing the ref_status' do
-        expect(PipelineNotificationWorker).to receive(:perform_async).with(pipeline.id, ref_status: pipeline.status)
-
-        subject.call
-      end
-    end
-
-    shared_examples 'updates ci_ref' do
-      where(:ref_status, :pipeline_status, :next_status) do
-        [
-          %w[failed success fixed],
-          %w[failed failed failed],
-          %w[success success success],
-          %w[success failed failed]
-        ]
-      end
-
-      with_them do
-        let(:ci_ref) { create(:ci_ref, status: ref_status) }
-        let(:pipeline) { create(:ci_pipeline, status: pipeline_status, project: ci_ref.project, ref: ci_ref.ref) }
-
-        it 'sets ci_ref.status to next_status' do
-          expect do
-            expect(subject.call).to eq(true)
-            expect(ci_ref.reload.status).to eq(next_status)
-          end.not_to change { Ci::Ref.count }
-        end
-
-        it 'calls PipelineNotificationWorker pasing the ref_status' do
-          expect(PipelineNotificationWorker).to receive(:perform_async).with(pipeline.id, ref_status: next_status)
-
-          subject.call
-        end
-      end
-    end
-
-    shared_examples 'does a noop' do
-      it "doesn't change ci_ref" do
-        expect do
-          expect do
-            expect(subject.call).to eq(false)
-          end.not_to change { ci_ref.reload.status }
-        end.not_to change { Ci::Ref.count }
-      end
-
-      it "doesn't call PipelineNotificationWorker" do
-        expect(PipelineNotificationWorker).not_to receive(:perform_async)
-
-        subject.call
-      end
-    end
-
-    context "ci_ref doesn't exists" do
-      let(:pipeline) { create(:ci_pipeline, :success, ref: 'new-ref') }
-
-      it_behaves_like 'creates ci_ref'
-
-      context 'when an ActiveRecord::RecordNotUnique validation is raised' do
-        let(:ci_ref) { create(:ci_ref, status: 'failed') }
-        let(:pipeline) { create(:ci_pipeline, status: :success, project: ci_ref.project, ref: ci_ref.ref) }
-
-        it 'reloads the ci_ref and retries once' do
-          subject.instance_variable_set("@ref", subject.send(:build_ref))
-
-          expect do
-            expect(subject.call).to eq(true)
-          end.not_to change { Ci::Ref.count }
-          expect(ci_ref.reload.status).to eq('fixed')
-        end
-
-        it 'raises error on multiple retries' do
-          allow_any_instance_of(Ci::Ref).to receive(:update)
-            .and_raise(ActiveRecord::RecordNotUnique)
-
-          expect { subject.call }.to raise_error(ActiveRecord::RecordNotUnique)
-        end
-      end
-    end
-
-    context 'ci_ref exists' do
-      let!(:ci_ref) { create(:ci_ref, status: 'failed') }
-      let(:pipeline) { ci_ref.pipelines.first }
-
-      it_behaves_like 'updates ci_ref'
-
-      context 'pipeline status is invalid' do
-        let!(:pipeline) { create(:ci_pipeline, :running, project: ci_ref.project, ref: ci_ref.ref, tag: ci_ref.tag) }
-
-        it_behaves_like 'does a noop'
-      end
-
-      context 'newer pipeline finished' do
-        let(:newer_pipeline) { create(:ci_pipeline, :success, project: ci_ref.project, ref: ci_ref.ref, tag: ci_ref.tag) }
-
-        before do
-          ci_ref.update!(last_updated_by_pipeline: newer_pipeline)
-        end
-
-        it_behaves_like 'does a noop'
-      end
-
-      context 'pipeline is retried' do
-        before do
-          ci_ref.update!(last_updated_by_pipeline: pipeline)
-        end
-
-        it_behaves_like 'updates ci_ref'
-      end
-
-      context 'ref is stale' do
-        let(:pipeline1) { create(:ci_pipeline, :success, project: ci_ref.project, ref: ci_ref.ref, tag: ci_ref.tag) }
-        let(:pipeline2) { create(:ci_pipeline, :success, project: ci_ref.project, ref: ci_ref.ref, tag: ci_ref.tag) }
-
-        it 'reloads the ref and retry' do
-          service1 = described_class.new(pipeline1)
-          service2 = described_class.new(pipeline2)
-
-          service2.send(:ref)
-          service1.call
-          expect(ci_ref.reload.status).to eq('fixed')
-          expect do
-            expect(service2.call).to eq(true)
-            # We expect 'success' in this case rather than 'fixed' because
-            # the ref is correctly reloaded on stale error.
-            expect(ci_ref.reload.status).to eq('success')
-          end.not_to change { Ci::Ref.count }
-        end
-
-        it 'aborts when a newer pipeline finished' do
-          service1 = described_class.new(pipeline1)
-          service2 = described_class.new(pipeline2)
-
-          service2.call
-          expect do
-            expect(service1.call).to eq(false)
-            expect(ci_ref.reload.status).to eq('fixed')
-          end.not_to change { Ci::Ref.count }
-        end
-      end
-
-      context 'ref exists as both tag/branch and tag' do
-        let(:pipeline) { create(:ci_pipeline, :failed, project: ci_ref.project, ref: ci_ref.ref, tag: true) }
-        let!(:branch_pipeline) { create(:ci_pipeline, :success, project: ci_ref.project, ref: ci_ref.ref, tag: false) }
-
-        it_behaves_like 'creates ci_ref'
-      end
-    end
-  end
-end

spec/workers/pipeline_update_ci_ref_status_worker_service_spec.rb

@@ -2,6 +2,7 @@
 
 require 'spec_helper'
 
+# NOTE: This class is unused and to be removed in 13.1~
 describe PipelineUpdateCiRefStatusWorker do
   let(:worker) { described_class.new }
   let(:pipeline) { create(:ci_pipeline) }