Ensure after_run ops execute once only

This change ensures any operations designed to execute
after a bulk_import pipeline has run will only be executed
once, whether the pipeline is batched or not batched. It
corrects the error where after_run operations were run
after each batch completion, rather than after all batches
were completed. The method/pipeline step has been named
'on_finish' for clarity.

Changelog: fixed

app/workers/all_queues.yml

@@ -3567,6 +3567,15 @@
   :weight: 1
   :idempotent: false
   :tags: []
+- :name: projects_import_export_after_import_merge_requests
+  :worker_name: Projects::ImportExport::AfterImportMergeRequestsWorker
+  :feature_category: :importers
+  :has_external_dependencies: false
+  :urgency: :low
+  :resource_boundary: :unknown
+  :weight: 1
+  :idempotent: true
+  :tags: []
 - :name: projects_import_export_create_relation_exports
   :worker_name: Projects::ImportExport::CreateRelationExportsWorker
   :feature_category: :importers

app/workers/bulk_imports/finish_batched_pipeline_worker.rb

@@ -16,6 +16,7 @@ class FinishBatchedPipelineWorker
 
     def perform(pipeline_tracker_id)
       @tracker = Tracker.find(pipeline_tracker_id)
+      @context = ::BulkImports::Pipeline::Context.new(tracker)
 
       return unless tracker.batched?
       return unless tracker.started?
@@ -26,6 +27,7 @@ def perform(pipeline_tracker_id)
         tracker.batches.map(&:fail_op!)
         tracker.fail_op!
       else
+        tracker.pipeline_class.new(@context).on_finish
         logger.info(log_attributes(message: 'Tracker finished'))
         tracker.finish!
       end

app/workers/projects/import_export/after_import_merge_requests_worker.rb

+# frozen_string_literal: true
+
+module Projects
+  module ImportExport
+    class AfterImportMergeRequestsWorker
+      include ApplicationWorker
+
+      idempotent!
+      data_consistency :delayed
+      urgency :low
+      feature_category :importers
+
+      def perform(project_id)
+        project = Project.find_by_id(project_id)
+        return unless project
+
+        project.merge_requests.set_latest_merge_request_diff_ids!
+      end
+    end
+  end
+end

config/sidekiq_queues.yml

@@ -563,6 +563,8 @@
   - 1
 - - projects_git_garbage_collect
   - 1
+- - projects_import_export_after_import_merge_requests
+  - 1
 - - projects_import_export_create_relation_exports
   - 1
 - - projects_import_export_parallel_project_export

lib/bulk_imports/pipeline/runner.rb

@@ -40,6 +40,14 @@ def run
           run_pipeline_step(:after_run) do
             after_run(extracted_data)
           end
+
+          # For batches, `#on_finish` is called once within `FinishBatchedPipelineWorker`
+          # after all batches have completed.
+          unless tracker.batched?
+            run_pipeline_step(:on_finish) do
+              on_finish
+            end
+          end
         end
 
         info(message: 'Pipeline finished')
@@ -47,6 +55,8 @@ def run
         skip!('Skipping pipeline due to failed entity')
       end
 
+      def on_finish; end
+
       private # rubocop:disable Lint/UselessAccessModifier
 
       def run_pipeline_step(step, class_name = nil, entry = nil)

lib/bulk_imports/projects/pipelines/merge_requests_pipeline.rb

@@ -10,8 +10,8 @@ class MergeRequestsPipeline
 
         extractor ::BulkImports::Common::Extractors::NdjsonExtractor, relation: relation
 
-        def after_run(_)
-          context.portable.merge_requests.set_latest_merge_request_diff_ids!
+        def on_finish
+          ::Projects::ImportExport::AfterImportMergeRequestsWorker.perform_async(context.portable.id)
         end
       end
     end

lib/bulk_imports/projects/pipelines/releases_pipeline.rb

@@ -10,9 +10,7 @@ class ReleasesPipeline
 
         extractor ::BulkImports::Common::Extractors::NdjsonExtractor, relation: relation
 
-        def after_run(_context)
-          super
-
+        def on_finish
           portable.releases.find_each do |release|
             create_release_evidence(release)
           end

spec/lib/bulk_imports/pipeline/runner_spec.rb

@@ -194,6 +194,8 @@ def load(context, data); end
             .with(context, extracted_data.data.first)
         end
 
+        expect(subject).to receive(:on_finish)
+
         expect_next_instance_of(Gitlab::Import::Logger) do |logger|
           expect(logger).to receive(:info)
             .with(
@@ -230,6 +232,14 @@ def load(context, data); end
                 step_class: 'BulkImports::Loader'
               )
             )
+          expect(logger).to receive(:info)
+            .with(
+              log_params(
+                context,
+                pipeline_class: 'BulkImports::MyPipeline',
+                pipeline_step: :on_finish
+              )
+            )
           expect(logger).to receive(:info)
             .with(
               log_params(
@@ -251,6 +261,28 @@ def load(context, data); end
         subject.run
       end
 
+      context 'when the pipeline is batched' do
+        let(:tracker) { create(:bulk_import_tracker, :batched, entity: entity) }
+
+        before do
+          allow_next_instance_of(BulkImports::Extractor) do |extractor|
+            allow(extractor).to receive(:extract).and_return(extracted_data)
+          end
+        end
+
+        it 'calls after_run' do
+          expect(subject).to receive(:after_run)
+
+          subject.run
+        end
+
+        it 'does not call on_finish' do
+          expect(subject).not_to receive(:on_finish)
+
+          subject.run
+        end
+      end
+
       context 'when extracted data has multiple pages' do
         it 'updates tracker information and runs pipeline again' do
           first_page = extracted_data(has_next_page: true)

spec/lib/bulk_imports/projects/pipelines/merge_requests_pipeline_spec.rb

@@ -2,7 +2,7 @@
 
 require 'spec_helper'
 
-RSpec.describe BulkImports::Projects::Pipelines::MergeRequestsPipeline do
+RSpec.describe BulkImports::Projects::Pipelines::MergeRequestsPipeline, feature_category: :importers do
   let_it_be(:user) { create(:user) }
   let_it_be(:another_user) { create(:user) }
   let_it_be(:group) { create(:group) }
@@ -43,6 +43,7 @@
         'base_commit_sha' => 'ae73cb07c9eeaf35924a10f713b364d32b2dd34f',
         'head_commit_sha' => 'a97f74ddaa848b707bea65441c903ae4bf5d844d',
         'start_commit_sha' => '9eea46b5c72ead701c22f516474b95049c9d9462',
+        'diff_type' => 1,
         'merge_request_diff_commits' => [
           {
             'sha' => 'COMMIT1',
@@ -99,6 +100,8 @@
       allow(project.repository).to receive(:branch_exists?).and_return(false)
       allow(project.repository).to receive(:create_branch)
 
+      allow(::Projects::ImportExport::AfterImportMergeRequestsWorker).to receive(:perform_async)
+
       pipeline.run
     end
 
@@ -244,8 +247,10 @@
         expect(imported_mr.merge_request_diff).to be_present
       end
 
-      it 'has the correct data for merge request latest_merge_request_diff' do
-        expect(imported_mr.latest_merge_request_diff_id).to eq(imported_mr.merge_request_diffs.maximum(:id))
+      it 'enqueues AfterImportMergeRequestsWorker worker' do
+        expect(::Projects::ImportExport::AfterImportMergeRequestsWorker)
+          .to have_received(:perform_async)
+          .with(project.id)
       end
 
       it 'imports diff files' do

spec/lib/bulk_imports/projects/pipelines/releases_pipeline_spec.rb

@@ -2,7 +2,7 @@
 
 require 'spec_helper'
 
-RSpec.describe BulkImports::Projects::Pipelines::ReleasesPipeline do
+RSpec.describe BulkImports::Projects::Pipelines::ReleasesPipeline, feature_category: :importers do
   let_it_be(:user) { create(:user) }
   let_it_be(:group) { create(:group) }
   let_it_be(:project) { create(:project, group: group) }

spec/workers/bulk_imports/finish_batched_pipeline_worker_spec.rb

@@ -5,16 +5,48 @@
 RSpec.describe BulkImports::FinishBatchedPipelineWorker, feature_category: :importers do
   let_it_be(:bulk_import) { create(:bulk_import) }
   let_it_be(:config) { create(:bulk_import_configuration, bulk_import: bulk_import) }
-  let_it_be(:entity) { create(:bulk_import_entity, bulk_import: bulk_import) }
+  let_it_be(:project) { create(:project, :repository) }
+  let_it_be(:entity) do
+    create(
+      :bulk_import_entity,
+      :project_entity,
+      project: project,
+      bulk_import: bulk_import
+    )
+  end
+
+  let(:pipeline_class) do
+    Class.new do
+      def initialize(_); end
+
+      def on_finish; end
+    end
+  end
 
-  let(:status_event) { :finish }
-  let(:pipeline_tracker) { create(:bulk_import_tracker, :started, :batched, entity: entity) }
+  let(:pipeline_tracker) do
+    create(
+      :bulk_import_tracker,
+      :started,
+      :batched,
+      entity: entity,
+      pipeline_name: 'FakePipeline'
+    )
+  end
 
   subject(:worker) { described_class.new }
 
   describe '#perform' do
+    before do
+      stub_const('FakePipeline', pipeline_class)
+
+      allow_next_instance_of(BulkImports::Projects::Stage) do |instance|
+        allow(instance).to receive(:pipelines)
+          .and_return([{ stage: 0, pipeline: pipeline_class }])
+      end
+    end
+
     context 'when import is in progress' do
-      it 'marks the pipeline as finished' do
+      it 'marks the tracker as finished' do
         expect_next_instance_of(BulkImports::Logger) do |logger|
           expect(logger).to receive(:info).with(
             a_hash_including('message' => 'Tracker finished')
@@ -26,6 +58,14 @@
           .from(false).to(true)
       end
 
+      it "calls the pipeline's `#on_finish`" do
+        expect_next_instance_of(pipeline_class) do |pipeline|
+          expect(pipeline).to receive(:on_finish)
+        end
+
+        subject.perform(pipeline_tracker.id)
+      end
+
       it 're-enqueues for any started batches' do
         create(:bulk_import_batch_tracker, :started, tracker: pipeline_tracker)
 
@@ -65,29 +105,39 @@
         expect(pipeline_tracker.batches.first.reload.failed?).to eq(true)
       end
     end
+  end
 
-    context 'when pipeline is not batched' do
-      let(:pipeline_tracker) { create(:bulk_import_tracker, :started, entity: entity) }
+  shared_examples 'does nothing' do
+    it "does not call the tracker's `#finish!`" do
+      expect_next_found_instance_of(BulkImports::Tracker) do |instance|
+        expect(instance).not_to receive(:finish!)
+      end
 
-      it 'returns' do
-        expect_next_instance_of(BulkImports::Tracker) do |instance|
-          expect(instance).not_to receive(:finish!)
-        end
+      subject.perform(pipeline_tracker.id)
+    end
 
-        subject.perform(pipeline_tracker.id)
-      end
+    it "does not call the pipeline's `#on_finish`" do
+      expect(pipeline_class).not_to receive(:new)
+
+      subject.perform(pipeline_tracker.id)
     end
+  end
 
-    context 'when pipeline is not started' do
-      let(:status_event) { :start }
+  context 'when tracker is not batched' do
+    let(:pipeline_tracker) { create(:bulk_import_tracker, :started, entity: entity, batched: false) }
 
-      it 'returns' do
-        expect_next_instance_of(BulkImports::Tracker) do |instance|
-          expect(instance).not_to receive(:finish!)
-        end
+    include_examples 'does nothing'
+  end
 
-        described_class.new.perform(pipeline_tracker.id)
-      end
-    end
+  context 'when tracker is not started' do
+    let(:pipeline_tracker) { create(:bulk_import_tracker, :batched, :finished, entity: entity) }
+
+    include_examples 'does nothing'
+  end
+
+  context 'when pipeline is enqueued' do
+    let(:pipeline_tracker) { create(:bulk_import_tracker, status: 3, entity: entity) }
+
+    include_examples 'does nothing'
   end
 end

spec/workers/projects/import_export/after_import_merge_requests_worker_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Projects::ImportExport::AfterImportMergeRequestsWorker, feature_category: :importers do
+  let_it_be(:project) { create(:project) }
+  let_it_be(:merge_requests) { project.merge_requests }
+
+  let(:worker) { described_class.new }
+
+  describe '#perform' do
+    it 'sets the latest merge request diff ids' do
+      expect(project.class).to receive(:find_by_id).and_return(project)
+      expect(merge_requests).to receive(:set_latest_merge_request_diff_ids!)
+
+      worker.perform(project.id)
+    end
+
+    it_behaves_like 'an idempotent worker' do
+      let(:job_args) { [project.id] }
+    end
+  end
+end