Refactor the logic of updating head pipelines

Sort out some logic

app/models/ci/pipeline.rb

@@ -178,6 +178,14 @@ module Ci
 
     scope :for_user, -> (user) { where(user: user) }
 
+    scope :for_merge_request, -> (merge_request, ref, sha) do
+      ##
+      # We have to filter out unrelated MR pipelines, in case,
+      # there are two merge requests from the same source branch
+      where(merge_request: [nil, merge_request], ref: ref, sha: sha)
+        .sort_by_merge_request_pipelines
+    end
+
     # Returns the pipelines in descending order (= newest first), optionally
     # limited to a number of references.
     #
@@ -265,6 +273,10 @@ module Ci
       sources.reject { |source| source == "external" }.values
     end
 
+    def self.latest_for_merge_request(merge_request, ref, sha)
+      for_merge_request(merge_request, ref, sha).first
+    end
+
     def self.ci_sources_values
       config_sources.values_at(:repository_source, :auto_devops_source, :unknown_source)
     end

app/models/merge_request.rb

@@ -1092,10 +1092,15 @@ class MergeRequest < ActiveRecord::Base
   def all_pipelines(shas: all_commit_shas)
     return Ci::Pipeline.none unless source_project
 
-    @all_pipelines ||= source_project.ci_pipelines
-      .where(sha: shas, ref: source_branch)
-      .where(merge_request: [nil, self])
-      .sort_by_merge_request_pipelines
+    @all_pipelines ||=
+      source_project.ci_pipelines
+                    .for_merge_request(self, source_branch, all_commit_shas)
+  end
+
+  def update_head_pipeline
+    self.head_pipeline = find_actual_head_pipeline
+
+    update_column(:head_pipeline_id, head_pipeline.id) if head_pipeline_id_changed?
   end
 
   def merge_request_pipeline_exists?
@@ -1295,6 +1300,11 @@ class MergeRequest < ActiveRecord::Base
       .find_by(sha: diff_base_sha)
   end
 
+  def find_actual_head_pipeline
+    source_project&.ci_pipelines
+                  &.latest_for_merge_request(self, source_branch, diff_head_sha)
+  end
+
   def discussions_rendered_on_frontend?
     true
   end

app/services/merge_requests/create_service.rb

@@ -26,7 +26,7 @@ module MergeRequests
       todo_service.new_merge_request(issuable, current_user)
       issuable.cache_merge_request_closes_issues!(current_user)
       create_merge_request_pipeline(issuable, current_user)
-      update_merge_requests_head_pipeline(issuable)
+      issuable.update_head_pipeline
 
       super
     end
@@ -45,20 +45,6 @@ module MergeRequests
 
     private
 
-    def update_merge_requests_head_pipeline(merge_request)
-      pipeline = head_pipeline_for(merge_request)
-      merge_request.update(head_pipeline_id: pipeline.id) if pipeline
-    end
-
-    def head_pipeline_for(merge_request)
-      return unless merge_request.source_project
-
-      sha = merge_request.source_branch_sha
-      return unless sha
-
-      merge_request.all_pipelines(shas: sha).first
-    end
-
     def set_projects!
       # @project is used to determine whether the user can set the merge request's
       # assignee, milestone and labels. Whether they can depends on their

app/workers/update_head_pipeline_for_merge_request_worker.rb

@@ -7,25 +7,8 @@ class UpdateHeadPipelineForMergeRequestWorker
   queue_namespace :pipeline_processing
 
   def perform(merge_request_id)
-    merge_request = MergeRequest.find(merge_request_id)
-
-    sha = merge_request.diff_head_sha
-    pipeline = merge_request.all_pipelines(shas: sha).first
-
-    return unless pipeline && pipeline.latest?
-
-    if merge_request.diff_head_sha != pipeline.sha
-      log_error_message_for(merge_request)
-
-      return
+    MergeRequest.find_by_id(merge_request_id).try do |merge_request|
+      merge_request.update_head_pipeline
     end
-
-    merge_request.update_attribute(:head_pipeline_id, pipeline.id)
-  end
-
-  def log_error_message_for(merge_request)
-    Rails.logger.error(
-      "Outdated head pipeline for active merge request: id=#{merge_request.id}, source_branch=#{merge_request.source_branch}, diff_head_sha=#{merge_request.diff_head_sha}"
-    )
   end
 end

changelogs/unreleased/user-update-head-pipeline-worker.yml

+---
+title: Refactor the logic of updating head pipelines for merge requests
+merge_request: 23437
+author:
+type: other

spec/models/merge_request_spec.rb

@@ -1372,6 +1372,34 @@ describe MergeRequest do
     end
   end
 
+  describe '#update_head_pipeline' do
+    subject { merge_request.update_head_pipeline }
+
+    let(:merge_request) { create(:merge_request) }
+
+    context 'when there is a pipeline with the diff head sha' do
+      let!(:pipeline) do
+        create(:ci_empty_pipeline,
+               project: merge_request.project,
+               sha: merge_request.diff_head_sha,
+               ref: merge_request.source_branch)
+      end
+
+      it 'updates the head pipeline' do
+        expect { subject }
+          .to change { merge_request.reload.head_pipeline }
+          .from(nil).to(pipeline)
+      end
+    end
+
+    context 'when there are no pipelines with the diff head sha' do
+      it 'does not update the head pipeline' do
+        expect { subject }
+          .not_to change { merge_request.reload.head_pipeline }
+      end
+    end
+  end
+
   describe '#has_test_reports?' do
     subject { merge_request.has_test_reports? }
 

spec/services/ci/create_pipeline_service_spec.rb

@@ -143,7 +143,8 @@ describe Ci::CreatePipelineService do
                                                    target_branch: "branch_1",
                                                    source_project: project)
 
-            allow_any_instance_of(Ci::Pipeline).to receive(:latest?).and_return(false)
+            allow_any_instance_of(MergeRequest)
+              .to receive(:find_actual_head_pipeline) { }
 
             execute_service
 

spec/services/merge_requests/create_service_spec.rb

@@ -128,9 +128,9 @@ describe MergeRequests::CreateService do
       end
 
       context 'when head pipelines already exist for merge request source branch' do
-        let(:sha) { project.commit(opts[:source_branch]).id }
-        let!(:pipeline_1) { create(:ci_pipeline, project: project, ref: opts[:source_branch], project_id: project.id, sha: sha) }
-        let!(:pipeline_2) { create(:ci_pipeline, project: project, ref: opts[:source_branch], project_id: project.id, sha: sha) }
+        let(:shas) { project.repository.commits(opts[:source_branch], limit: 2).map(&:id) }
+        let!(:pipeline_1) { create(:ci_pipeline, project: project, ref: opts[:source_branch], project_id: project.id, sha: shas[1]) }
+        let!(:pipeline_2) { create(:ci_pipeline, project: project, ref: opts[:source_branch], project_id: project.id, sha: shas[0]) }
         let!(:pipeline_3) { create(:ci_pipeline, project: project, ref: "other_branch", project_id: project.id) }
 
         before do
@@ -144,17 +144,30 @@ describe MergeRequests::CreateService do
         it 'sets head pipeline' do
           merge_request = service.execute
 
-          expect(merge_request.head_pipeline).to eq(pipeline_2)
+          expect(merge_request.reload.head_pipeline).to eq(pipeline_2)
           expect(merge_request).to be_persisted
         end
 
-        context 'when merge request head commit sha does not match pipeline sha' do
-          it 'sets the head pipeline correctly' do
-            pipeline_2.update(sha: 1234)
+        context 'when the new pipeline is associated with an old sha' do
+          let!(:pipeline_1) { create(:ci_pipeline, project: project, ref: opts[:source_branch], project_id: project.id, sha: shas[0]) }
+          let!(:pipeline_2) { create(:ci_pipeline, project: project, ref: opts[:source_branch], project_id: project.id, sha: shas[1]) }
 
+          it 'sets an old pipeline with associated with the latest sha as the head pipeline' do
             merge_request = service.execute
 
-            expect(merge_request.head_pipeline).to eq(pipeline_1)
+            expect(merge_request.reload.head_pipeline).to eq(pipeline_1)
+            expect(merge_request).to be_persisted
+          end
+        end
+
+        context 'when there are no pipelines with the diff head sha' do
+          let!(:pipeline_1) { create(:ci_pipeline, project: project, ref: opts[:source_branch], project_id: project.id, sha: shas[1]) }
+          let!(:pipeline_2) { create(:ci_pipeline, project: project, ref: opts[:source_branch], project_id: project.id, sha: shas[1]) }
+
+          it 'does not set the head pipeline' do
+            merge_request = service.execute
+
+            expect(merge_request.reload.head_pipeline).to be_nil
             expect(merge_request).to be_persisted
           end
         end

spec/workers/update_head_pipeline_for_merge_request_worker_spec.rb

@@ -21,17 +21,17 @@ describe UpdateHeadPipelineForMergeRequestWorker do
           merge_request.merge_request_diff.update(head_commit_sha: 'different_sha')
         end
 
-        it 'does not update head_pipeline_id' do
-          expect { subject.perform(merge_request.id) }.not_to raise_error
-
-          expect(merge_request.reload.head_pipeline_id).to eq(nil)
+        it 'does not update head pipeline' do
+          expect { subject.perform(merge_request.id) }
+            .not_to change { merge_request.reload.head_pipeline_id }
         end
       end
     end
 
     context 'when pipeline does not exist for the source project and branch' do
       it 'does not update the head_pipeline_id of the merge_request' do
-        expect { subject.perform(merge_request.id) }.not_to change { merge_request.reload.head_pipeline_id }
+        expect { subject.perform(merge_request.id) }
+          .not_to change { merge_request.reload.head_pipeline_id }
       end
     end