Create refs/train and run a pipeline on it

Currently, pipelines for merge trains run on refs/merge paths,
however, this doesn't allow us to run pipeline in parallel by chaning
merge refs, so we'll introduce a refs/train for this specific purpose.

ee/app/models/merge_train.rb

@@ -41,12 +41,20 @@ class MergeTrain < ApplicationRecord
     self.class.all_in_train(merge_request).where('merge_trains.id > ?', id)
   end
 
+  def all_prev
+    self.class.all_in_train(merge_request).where('merge_trains.id < ?', id)
+  end
+
   def next
     all_next.first
   end
 
+  def prev
+    all_prev.last
+  end
+
   def index
-    self.class.all_in_train(merge_request).where('merge_trains.id < ?', id).count
+    all_prev.count
   end
 
   def first_in_train?
@@ -54,6 +62,6 @@ class MergeTrain < ApplicationRecord
   end
 
   def follower_in_train?
-    self.class.all_in_train(merge_request).where('merge_trains.id < ?', id).exists?
+    all_prev.exists?
   end
 end

ee/app/services/merge_trains/create_pipeline_service.rb

 # frozen_string_literal: true
 module MergeTrains
   class CreatePipelineService < BaseService
-    def execute(merge_request)
+    def execute(merge_request, previous_ref)
       validation_status = validate(merge_request)
       return validation_status unless validation_status[:status] == :success
 
-      merge_status = create_merge_ref(merge_request)
+      merge_status = create_train_ref(merge_request, previous_ref)
       return error(merge_status[:message]) unless merge_status[:status] == :success
 
       create_pipeline(merge_request, merge_status)
@@ -21,11 +21,23 @@ module MergeTrains
       success
     end
 
-    def create_merge_ref(merge_request)
-      ::MergeRequests::MergeToRefService.new(merge_request.project, merge_request.merge_user, target_ref: merge_request.train_ref_path)
+    def create_train_ref(merge_request, previous_ref)
+      return error('previous ref is not specified') unless previous_ref
+
+      commit_message = commit_message(merge_request, previous_ref)
+
+      ::MergeRequests::MergeToRefService.new(merge_request.project, merge_request.merge_user,
+                                             target_ref: merge_request.train_ref_path,
+                                             first_parent_ref: previous_ref,
+                                             commit_message: commit_message)
                                         .execute(merge_request)
     end
 
+    def commit_message(merge_request, previous_ref)
+      "Merge branch #{merge_request.source_branch} with #{previous_ref} " \
+      "into #{merge_request.train_ref_path}"
+    end
+
     def create_pipeline(merge_request, merge_status)
       pipeline = ::Ci::CreatePipelineService.new(merge_request.source_project, merge_request.merge_user,
         ref: merge_request.train_ref_path,

ee/app/services/merge_trains/refresh_merge_request_service.rb

@@ -15,10 +15,10 @@ module MergeTrains
 
       validate!
 
-      create_pipeline! if should_create_pipeline?
+      pipeline_created = create_pipeline! if should_create_pipeline?
       merge! if should_merge?
 
-      success
+      success(pipeline_created: pipeline_created.present?)
     rescue ProcessError => e
       drop(e)
     end
@@ -34,10 +34,18 @@ module MergeTrains
         raise ProcessError, 'merge request is not on a merge train'
       end
 
+      if merge_request.squash?
+        raise ProcessError, 'merge train does not support squash merge'
+      end
+
       unless merge_request.mergeable_state?(skip_ci_check: true)
         raise ProcessError, 'merge request is not mergeable'
       end
 
+      unless previous_ref_exist?
+        raise ProcessError, 'previous ref does not exist'
+      end
+
       if pipeline_for_merge_train
         if pipeline_for_merge_train.complete? && !pipeline_for_merge_train.success?
           raise ProcessError, 'pipeline did not succeed'
@@ -45,13 +53,16 @@ module MergeTrains
       end
     end
 
+    # Since `stale_pipeline?` is expensive process which requires multiple Gitaly calls,
+    # each refresh service relays `require_recreate` flag whether the next
+    # merge request obviously requires to re-create pipeline for merge train.
     def should_create_pipeline?
-      first_in_train? && (pipeline_absent? || stale_pipeline?)
+      pipeline_absent? || require_recreate? || stale_pipeline?
     end
 
     def create_pipeline!
       result = MergeTrains::CreatePipelineService.new(merge_request.project, merge_user)
-        .execute(merge_request)
+        .execute(merge_request, previous_ref)
 
       raise ProcessError, result[:message] unless result[:status] == :success
 
@@ -71,8 +82,23 @@ module MergeTrains
       merge_train.destroy
     end
 
+    # NOTE: This method works for both no-ff-merge and ff-merge, however,
+    #       it doesn't work for squash and merge option.
     def stale_pipeline?
-      pipeline_for_merge_train && !pipeline_for_merge_train.latest_merge_request_pipeline?
+      return true unless pipeline_for_merge_train.source_sha == merge_request.diff_head_sha
+      return false if pipeline_for_merge_train.target_sha == previous_ref_sha
+
+      ##
+      # Now `pipeline.target_sha` and `previous_ref_sha` are different. This case
+      # happens in the following cases:
+      # 1. Previous sha has a completely different history from the pipeline.target_sha.
+      #    e.g. Previous merge request was dropped from the merge train.
+      # 2. Previous sha has exactly the same history with the pipeline.target_sha.
+      #    e.g. Previous merge request was merged into target branch with no-ff option.
+      #
+      # We distinguish these two cases by comparing parent commits.
+      commits = merge_request.project.commits_by(oids: [pipeline_for_merge_train.target_sha, previous_ref_sha])
+      commits[0].parent_ids != commits[1].parent_ids
     end
 
     def pipeline_absent?
@@ -97,6 +123,30 @@ module MergeTrains
       end
     end
 
+    def previous_ref_sha
+      strong_memoize(:previous_ref_sha) do
+        merge_request.project.repository.commit(previous_ref)&.sha
+      end
+    end
+
+    def previous_ref
+      previous_merge_request&.train_ref_path || merge_request.target_branch_ref
+    end
+
+    def previous_ref_exist?
+      previous_ref_sha.present?
+    end
+
+    def previous_merge_request
+      strong_memoize(:previous_merge_request) do
+        merge_request.merge_train.prev
+      end
+    end
+
+    def require_recreate?
+      params[:require_recreate]
+    end
+
     def drop(error)
       AutoMerge::MergeTrainService.new(project, merge_user)
         .abort(merge_request, error.message)

ee/app/services/merge_trains/refresh_merge_requests_service.rb

@@ -2,6 +2,10 @@
 module MergeTrains
   class RefreshMergeRequestsService < BaseService
     include ::Gitlab::ExclusiveLeaseHelpers
+    include ::Gitlab::Utils::StrongMemoize
+
+    DEFAULT_MAX_CONCURRENCY = 4
+    ONE_AT_A_TIME_STRATEGY = 1
 
     ##
     # merge_request ... A merge request pointer in a merge train.
@@ -39,10 +43,17 @@ module MergeTrains
     end
 
     def unsafe_refresh(merge_request)
+      require_next_recreate = false
+
       following_merge_requests_from(merge_request).each do |merge_request|
-        MergeTrains::RefreshMergeRequestService
-          .new(merge_request.project, merge_request.merge_user)
+        break if merge_request.merge_train.index >= max_concurrency
+
+        result = MergeTrains::RefreshMergeRequestService
+          .new(merge_request.project, merge_request.merge_user,
+               require_recreate: require_next_recreate)
           .execute(merge_request)
+
+        require_next_recreate = (result[:status] == :error || result[:pipeline_created])
       end
     end
 
@@ -53,5 +64,15 @@ module MergeTrains
     def queue_id(merge_request)
       "#{merge_request.target_project_id}:#{merge_request.target_branch}"
     end
+
+    def max_concurrency
+      strong_memoize(:max_concurrency) do
+        if Feature.enabled?(:merge_trains_parallel_pipelines, project, default_enabled: true)
+          DEFAULT_MAX_CONCURRENCY
+        else
+          ONE_AT_A_TIME_STRATEGY
+        end
+      end
+    end
   end
 end

ee/changelogs/unreleased/run-pipeline-for-merge-trains-on-train-ref.yml

+---
+title: Build cascading train refs for parallel execution of Pipelines for merge trains
+merge_request: 14296
+author:
+type: added

ee/spec/features/merge_trains/two_merge_requests_on_train_spec.rb

@@ -3,6 +3,8 @@
 require 'rails_helper'
 
 describe 'Two merge requests on a merge train' do
+  include RepoHelpers
+
   let(:project) { create(:project, :repository) }
   set(:maintainer_1) { create(:user) }
   set(:maintainer_2) { create(:user) }
@@ -49,10 +51,17 @@ describe 'Two merge requests on a merge train' do
   it 'creates a pipeline for merge request 1' do
     expect(merge_request_1.merge_train.pipeline).to be_merge_request_pipeline
     expect(merge_request_1.merge_train.pipeline.user).to eq(maintainer_1)
+    expect(merge_request_1.merge_train.pipeline.ref).to eq(merge_request_1.train_ref_path)
+    expect(merge_request_1.merge_train.pipeline.target_sha)
+      .to eq(project.repository.commit('refs/heads/master').sha)
   end
 
-  it 'does not create a pipeline for merge request 2' do
-    expect(merge_request_2.merge_train.pipeline).to be_nil
+  it 'creates a pipeline for merge request 2' do
+    expect(merge_request_2.merge_train.pipeline).to be_merge_request_pipeline
+    expect(merge_request_2.merge_train.pipeline.user).to eq(maintainer_2)
+    expect(merge_request_2.merge_train.pipeline.ref).to eq(merge_request_2.train_ref_path)
+    expect(merge_request_2.merge_train.pipeline.target_sha)
+      .to eq(project.repository.commit(merge_request_1.train_ref_path).sha)
   end
 
   it 'does not merge anything yet' do
@@ -60,26 +69,39 @@ describe 'Two merge requests on a merge train' do
     expect(merge_request_2).to be_opened
   end
 
-  context 'when the pipeline for merge request 1 succeeded' do
-    before do
-      merge_request_1.merge_train.pipeline.succeed!
-
-      merge_request_1.reload
-      merge_request_2.reload
+  shared_examples_for 'drops merge request 1 from the merge train' do
+    it 'drops merge request 1 from the merge train' do
+      expect(merge_request_1).to be_opened
+      expect(merge_request_1.merge_train).to be_nil
+      expect(merge_request_1.notes.last.note).to eq(system_note)
     end
+  end
 
-    it 'merges merge request 1' do
-      expect(merge_request_1).to be_merged
-      expect(merge_request_1.metrics.merged_by).to eq(maintainer_1)
+  shared_examples_for 'has an intact pipeline for merge request 2' do
+    it 'does not create a new pipeline for merge request 2' do
+      expect(merge_request_2.all_pipelines.count).to eq(1)
     end
 
-    it 'removes merge request 1 from the merge train' do
-      expect(merge_request_1.merge_train).to be_nil
+    context 'when the pipeline for merge request 2 succeeded' do
+      before do
+        merge_request_2.merge_train.pipeline.succeed!
+
+        merge_request_2.reload
+      end
+
+      it 'merges merge request 2' do
+        expect(merge_request_2).to be_merged
+        expect(merge_request_2.metrics.merged_by).to eq(maintainer_2)
+        expect(merge_request_2.merge_train).to be_nil
+      end
     end
+  end
 
-    it 'creates a pipeline for merge request 2' do
-      expect(merge_request_2.merge_train.pipeline).to be_merge_request_pipeline
-      expect(merge_request_2.merge_train.pipeline.user).to eq(maintainer_2)
+  shared_examples_for 're-creates a pipeline for merge request 2' do
+    it 'has recreated pipeline' do
+      expect(merge_request_2.all_pipelines.count).to eq(2)
+      expect(merge_request_2.merge_train.pipeline.target_sha)
+        .to eq(target_branch_sha)
     end
 
     context 'when the pipeline for merge request 2 succeeded' do
@@ -92,34 +114,44 @@ describe 'Two merge requests on a merge train' do
       it 'merges merge request 2' do
         expect(merge_request_2).to be_merged
         expect(merge_request_2.metrics.merged_by).to eq(maintainer_2)
-      end
-
-      it 'removes merge request 2 from the merge train' do
         expect(merge_request_2.merge_train).to be_nil
       end
     end
   end
 
-  context 'when the pipeline for merge request 1 failed' do
+  context 'when the pipeline for merge request 1 succeeded' do
     before do
-      merge_request_1.merge_train.pipeline.drop!
+      merge_request_1.merge_train.pipeline.succeed!
 
       merge_request_1.reload
       merge_request_2.reload
     end
 
-    it 'does not merges merge request 1' do
-      expect(merge_request_1).to be_opened
+    it 'merges merge request 1' do
+      expect(merge_request_1).to be_merged
+      expect(merge_request_1.metrics.merged_by).to eq(maintainer_1)
+      expect(merge_request_1.merge_train).to be_nil
     end
 
-    it 'drops merge request 1 from the merge train' do
-      expect(merge_request_1.merge_train).to be_nil
-      expect(merge_request_1.notes.last.note).to eq('removed this merge request from the merge train because pipeline did not succeed')
+    it_behaves_like 'has an intact pipeline for merge request 2'
+  end
+
+  context 'when the pipeline for merge request 1 failed' do
+    before do
+      merge_request_1.merge_train.pipeline.drop!
+
+      merge_request_1.reload
+      merge_request_2.reload
+    end
+
+    it_behaves_like 'drops merge request 1 from the merge train' do
+      let(:system_note) do
+        'removed this merge request from the merge train because pipeline did not succeed'
+      end
     end
 
-    it 'creates a pipeline for merge request 2' do
-      expect(merge_request_2.merge_train.pipeline).to be_merge_request_pipeline
-      expect(merge_request_2.merge_train.pipeline.user).to eq(maintainer_2)
+    it_behaves_like 're-creates a pipeline for merge request 2' do
+      let(:target_branch_sha) { project.repository.commit('refs/heads/master').sha }
     end
   end
 
@@ -131,14 +163,14 @@ describe 'Two merge requests on a merge train' do
       merge_request_2.reload
     end
 
-    it 'drops merge request 1 from the merge train' do
-      expect(merge_request_1.merge_train).to be_nil
-      expect(merge_request_1.notes.last.note).to eq('removed this merge request from the merge train')
+    it_behaves_like 'drops merge request 1 from the merge train' do
+      let(:system_note) do
+        'removed this merge request from the merge train'
+      end
     end
 
-    it 'creates a pipeline for merge request 2' do
-      expect(merge_request_2.merge_train.pipeline).to be_merge_request_pipeline
-      expect(merge_request_2.merge_train.pipeline.user).to eq(maintainer_2)
+    it_behaves_like 're-creates a pipeline for merge request 2' do
+      let(:target_branch_sha) { project.repository.commit('refs/heads/master').sha }
     end
   end
 
@@ -151,14 +183,62 @@ describe 'Two merge requests on a merge train' do
       merge_request_2.reload
     end
 
-    it 'drops merge request 1 from the merge train' do
-      expect(merge_request_1.merge_train).to be_nil
-      expect(merge_request_1.notes.last.note).to eq('removed this merge request from the merge train because merge request is not mergeable')
+    it_behaves_like 'drops merge request 1 from the merge train' do
+      let(:system_note) do
+        'removed this merge request from the merge train because merge request is not mergeable'
+      end
+    end
+
+    it_behaves_like 're-creates a pipeline for merge request 2' do
+      let(:target_branch_sha) { project.repository.commit('refs/heads/master').sha }
+    end
+  end
+
+  context 'when master got a new commit and pipeline for merge request 1 finished' do
+    before do
+      create_file_in_repo(project, 'master', 'master', 'test.txt', 'This is test')
+
+      merge_request_1.merge_train.pipeline.succeed!
+      merge_request_1.reload
+      merge_request_2.reload
+    end
+
+    it 're-creates a pipeline for merge request 1' do
+      expect(merge_request_1.all_pipelines.count).to eq(2)
+      expect(merge_request_1.merge_train.pipeline.target_sha)
+        .to eq(merge_request_1.target_branch_sha)
+    end
+
+    it 're-creates a pipeline for merge request 2' do
+      expect(merge_request_2.all_pipelines.count).to eq(2)
+      expect(merge_request_2.merge_train.pipeline.target_sha)
+        .to eq(project.repository.commit(merge_request_1.train_ref_path).sha)
     end
 
-    it 'creates a pipeline for merge request 2' do
-      expect(merge_request_2.merge_train.pipeline).to be_merge_request_pipeline
-      expect(merge_request_2.merge_train.pipeline.user).to eq(maintainer_2)
+    context 'when the pipeline for merge request 1 succeeded' do
+      before do
+        merge_request_1.merge_train.pipeline.succeed!
+
+        merge_request_1.reload
+      end
+
+      it 'merges merge request 1' do
+        expect(merge_request_1).to be_merged
+        expect(merge_request_1.metrics.merged_by).to eq(maintainer_1)
+      end
+
+      context 'when the pipeline for merge request 2 succeeded' do
+        before do
+          merge_request_2.merge_train.pipeline.succeed!
+
+          merge_request_2.reload
+        end
+
+        it 'merges merge request 2' do
+          expect(merge_request_2).to be_merged
+          expect(merge_request_2.metrics.merged_by).to eq(maintainer_2)
+        end
+      end
     end
   end
 

ee/spec/models/merge_train_spec.rb

@@ -142,6 +142,28 @@ describe MergeTrain do
     end
   end
 
+  describe '#all_prev' do
+    subject { merge_train.all_prev }
+
+    let(:merge_train) { merge_request.merge_train }
+    let!(:merge_request) { create_merge_request_on_train }
+
+    context 'when the merge request is at first on the train' do
+      it 'returns nil' do
+        is_expected.to be_empty
+      end
+    end
+
+    context 'when the merge request is at last on the train' do
+      let(:merge_train) { merge_request_2.merge_train }
+      let!(:merge_request_2) { create_merge_request_on_train(source_branch: 'improve/awesome') }
+
+      it 'returns the previous merge requests' do
+        is_expected.to eq([merge_request])
+      end
+    end
+  end
+
   describe '#next' do
     subject { merge_train.next }
 
@@ -163,6 +185,28 @@ describe MergeTrain do
     end
   end
 
+  describe '#prev' do
+    subject { merge_train.prev }
+
+    let(:merge_train) { merge_request.merge_train }
+    let!(:merge_request) { create_merge_request_on_train }
+
+    context 'when the merge request is at first on the train' do
+      it 'returns nil' do
+        is_expected.to be_nil
+      end
+    end
+
+    context 'when the merge request is at last on the train' do
+      let(:merge_train) { merge_request_2.merge_train }
+      let!(:merge_request_2) { create_merge_request_on_train(source_branch: 'improve/awesome') }
+
+      it 'returns the next merge request' do
+        is_expected.to eq(merge_request)
+      end
+    end
+  end
+
   describe '#first_in_train?' do
     subject { merge_train.first_in_train? }
 

ee/spec/services/merge_trains/create_pipeline_service_spec.rb

@@ -3,9 +3,10 @@
 require 'spec_helper'
 
 describe MergeTrains::CreatePipelineService do
-  set(:project) { create(:project, :repository) }
+  let(:project) { create(:project, :repository) }
   set(:maintainer) { create(:user) }
   let(:service) { described_class.new(project, maintainer) }
+  let(:previous_ref) { 'refs/heads/master' }
 
   before do
     project.add_maintainer(maintainer)
@@ -14,7 +15,7 @@ describe MergeTrains::CreatePipelineService do
   end
 
   describe '#execute' do
-    subject { service.execute(merge_request) }
+    subject { service.execute(merge_request, previous_ref) }
 
     let!(:merge_request) do
       create(:merge_request, :on_train, train_creator: maintainer,
@@ -82,6 +83,10 @@ describe MergeTrains::CreatePipelineService do
           expect { subject }
             .to change { merge_request.project.repository.ref_exists?(merge_request.train_ref_path) }
             .from(false).to(true)
+
+          expect(project.repository.commit(merge_request.train_ref_path).message)
+            .to eq("Merge branch #{merge_request.source_branch} with #{previous_ref} " \
+                   "into #{merge_request.train_ref_path}")
         end
 
         it 'calls Ci::CreatePipelineService for creating pipeline on train ref' do
@@ -92,6 +97,39 @@ describe MergeTrains::CreatePipelineService do
 
           subject
         end
+
+        context 'when previous_ref is a train ref' do
+          let(:previous_ref) { 'refs/merge-requests/999/train' }
+          let(:previous_ref_sha) { project.repository.commit('refs/merge-requests/999/train').sha }
+
+          context 'when previous_ref exists' do
+            before do
+              project.repository.create_ref('master', previous_ref)
+            end
+
+            it 'creates train ref with the specified ref' do
+              subject
+
+              commit = project.repository.commit(merge_request.train_ref_path)
+              expect(commit.parent_ids[1]).to eq(merge_request.diff_head_sha)
+              expect(commit.parent_ids[0]).to eq(previous_ref_sha)
+            end
+          end
+
+          context 'when previous_ref does not exist' do
+            it_behaves_like 'returns an error' do
+              let(:expected_reason) { '3:Invalid merge source' }
+            end
+          end
+        end
+
+        context 'when previous_ref is nil' do
+          let(:previous_ref) { nil }
+
+          it_behaves_like 'returns an error' do
+            let(:expected_reason) { 'previous ref is not specified' }
+          end
+        end
       end
 
       context 'when .gitlab-ci.yml does not have only: [merge_requests] specification' do

ee/spec/services/merge_trains/refresh_merge_request_service_spec.rb

@@ -5,7 +5,8 @@ require 'spec_helper'
 describe MergeTrains::RefreshMergeRequestService do
   set(:project) { create(:project, :repository) }
   set(:maintainer) { create(:user) }
-  let(:service) { described_class.new(project, maintainer) }
+  let(:service) { described_class.new(project, maintainer, require_recreate: require_recreate) }
+  let(:require_recreate) { false }
 
   before do
     project.add_maintainer(maintainer)
@@ -34,6 +35,31 @@ describe MergeTrains::RefreshMergeRequestService do
       end
     end
 
+    shared_examples_for 'creates a pipeline for merge train' do
+      let(:previous_ref) { 'refs/heads/master' }
+
+      it do
+        expect_next_instance_of(MergeTrains::CreatePipelineService, project, maintainer) do |pipeline_service|
+          allow(pipeline_service).to receive(:execute) { { status: :success, pipeline: pipeline } }
+          expect(pipeline_service).to receive(:execute).with(merge_request, previous_ref)
+        end
+
+        result = subject
+        expect(result[:status]).to eq(:success)
+        expect(result[:pipeline_created]).to eq(true)
+      end
+    end
+
+    shared_examples_for 'does not create a pipeline' do
+      it do
+        expect(service).not_to receive(:create_pipeline!)
+
+        result = subject
+        expect(result[:status]).to eq(:success)
+        expect(result[:pipeline_created]).to be_falsy
+      end
+    end
+
     context 'when merge train project configuration is disabled' do
       before do
         project.update!(merge_trains_enabled: false)
@@ -70,15 +96,33 @@ describe MergeTrains::RefreshMergeRequestService do
       end
     end
 
+    context 'when merge request is to be squashed' do
+      before do
+        merge_request.update!(squash: true)
+      end
+
+      it_behaves_like 'drops the merge request from the merge train' do
+        let(:expected_reason) { 'merge train does not support squash merge' }
+      end
+    end
+
+    context 'when previous ref is not found' do
+      let(:previous_ref) { 'refs/tmp/test' }
+
+      before do
+        allow(service).to receive(:previous_ref) { previous_ref }
+      end
+
+      it_behaves_like 'drops the merge request from the merge train' do
+        let(:expected_reason) { 'previous ref does not exist' }
+      end
+    end
+
     context 'when pipeline has not been created yet' do
       context 'when the merge request is the first queue' do
-        it 'creates a pipeline for merge train' do
-          expect_next_instance_of(MergeTrains::CreatePipelineService, project, maintainer) do |pipeline_service|
-            expect(pipeline_service).to receive(:execute).with(merge_request).and_call_original
-          end
+        let(:pipeline) { create(:ci_pipeline) }
 
-          subject
-        end
+        it_behaves_like 'creates a pipeline for merge train'
 
         context 'when it failed to create a pipeline' do
           before do
@@ -90,25 +134,38 @@ describe MergeTrains::RefreshMergeRequestService do
           end
         end
       end
+    end
 
-      context 'when the merge request is not the first queue' do
-        before do
-          allow(merge_request.merge_train).to receive(:first_in_train?) { false }
-        end
+    context 'when pipeline for merge train is running' do
+      let(:pipeline) { create(:ci_pipeline, :running, target_sha: previous_ref_sha, source_sha: merge_request.diff_head_sha) }
+      let(:previous_ref_sha) { project.repository.commit('refs/heads/master').sha }
 
-        it 'does not create a pipeline for merge train' do
-          expect(MergeTrains::CreatePipelineService).not_to receive(:new)
+      before do
+        merge_request.merge_train.update!(pipeline: pipeline)
+      end
 
-          subject
-        end
+      context 'when the pipeline is not stale' do
+        it_behaves_like 'does not create a pipeline'
+      end
+
+      context 'when the pipeline is stale' do
+        let(:previous_ref_sha) { project.repository.commits('refs/heads/master', limit: 2).last.sha }
+
+        it_behaves_like 'creates a pipeline for merge train'
+      end
+
+      context 'when the pipeline is reuired to be recreated' do
+        let(:require_recreate) { true }
+
+        it_behaves_like 'creates a pipeline for merge train'
       end
     end
 
     context 'when pipeline for merge train succeeded' do
-      let(:pipeline) { create(:ci_pipeline, :success) }
+      let(:pipeline) { create(:ci_pipeline, :success, target_sha: previous_ref_sha, source_sha: merge_request.diff_head_sha) }
+      let(:previous_ref_sha) { project.repository.commit('refs/heads/master').sha }
 
       before do
-        allow(pipeline).to receive(:latest_merge_request_pipeline?) { true }
         merge_request.merge_train.update!(pipeline: pipeline)
       end
 

ee/spec/services/merge_trains/refresh_merge_requests_service_spec.rb

@@ -15,7 +15,7 @@ describe MergeTrains::RefreshMergeRequestsService do
     project.add_maintainer(maintainer_2)
   end
 
-  describe '#execute' do
+  describe '#execute', :clean_gitlab_redis_queues do
     subject { service.execute(merge_request) }
 
     let!(:merge_request_1) do
@@ -32,12 +32,17 @@ describe MergeTrains::RefreshMergeRequestsService do
 
     let(:refresh_service_1) { double }
     let(:refresh_service_2) { double }
+    let(:refresh_service_1_result) { { status: :success } }
+    let(:refresh_service_2_result) { { status: :success } }
 
     before do
       allow(MergeTrains::RefreshMergeRequestService)
-        .to receive(:new).with(project, maintainer_1) { refresh_service_1 }
+        .to receive(:new).with(project, maintainer_1, anything) { refresh_service_1 }
       allow(MergeTrains::RefreshMergeRequestService)
-        .to receive(:new).with(project, maintainer_2) { refresh_service_2 }
+        .to receive(:new).with(project, maintainer_2, anything) { refresh_service_2 }
+
+      allow(refresh_service_1).to receive(:execute) { refresh_service_1_result }
+      allow(refresh_service_2).to receive(:execute) { refresh_service_2_result }
     end
 
     context 'when merge request 1 is passed' do
@@ -50,6 +55,52 @@ describe MergeTrains::RefreshMergeRequestsService do
         subject
       end
 
+      context 'when merge_trains_parallel_pipelines feature flag is disabled' do
+        before do
+          stub_feature_flags(merge_trains_parallel_pipelines: false)
+        end
+
+        it 'does not refresh merge request 2' do
+          expect(refresh_service_1).to receive(:execute).with(merge_request_1)
+          expect(refresh_service_2).not_to receive(:execute).with(merge_request_2)
+
+          subject
+        end
+      end
+
+      context 'when refresh service 1 returns error status' do
+        let(:refresh_service_1_result) { { status: :error, message: 'Failed to create ref' } }
+
+        it 'specifies require_recreate to refresh service 2' do
+          allow(MergeTrains::RefreshMergeRequestService)
+            .to receive(:new).with(project, maintainer_2, require_recreate: true) { refresh_service_2 }
+
+          subject
+        end
+      end
+
+      context 'when refresh service 1 returns success status and did not create a pipeline' do
+        let(:refresh_service_1_result) { { status: :success, pipeline_created: false } }
+
+        it 'does not specify require_recreate to refresh service 2' do
+          allow(MergeTrains::RefreshMergeRequestService)
+            .to receive(:new).with(project, maintainer_2, require_recreate: false) { refresh_service_2 }
+
+          subject
+        end
+      end
+
+      context 'when refresh service 1 returns success status and created a pipeline' do
+        let(:refresh_service_1_result) { { status: :success, pipeline_created: true } }
+
+        it 'specifies require_recreate to refresh service 2' do
+          allow(MergeTrains::RefreshMergeRequestService)
+            .to receive(:new).with(project, maintainer_2, require_recreate: true) { refresh_service_2 }
+
+          subject
+        end
+      end
+
       context 'when merge request 1 is not on a merge train' do
         let(:merge_request) { merge_request_1 }
         let!(:merge_request_1) { create(:merge_request) }