Rename CreateCrossProjectPipeline worker

Renames the CreateCrossProjectPipeline worker to
CreateDownstreamPipeline.

app/models/ci/bridge.rb

@@ -31,7 +31,11 @@ module Ci
         next unless bridge.triggers_downstream_pipeline?
 
         bridge.run_after_commit do
-          ::Ci::CreateCrossProjectPipelineWorker.perform_async(bridge.id)
+          if ::Feature.enabled?(:create_cross_project_pipeline_worker_rename, default_enabled: :yaml)
+            ::Ci::CreateDownstreamPipelineWorker.perform_async(bridge.id)
+          else
+            ::Ci::CreateCrossProjectPipelineWorker.perform_async(bridge.id)
+          end
         end
       end
 

app/workers/all_queues.yml

@@ -1474,6 +1474,15 @@
   :weight: 3
   :idempotent:
   :tags: []
+- :name: pipeline_default:ci_create_downstream_pipeline
+  :worker_name: Ci::CreateDownstreamPipelineWorker
+  :feature_category: :continuous_integration
+  :has_external_dependencies:
+  :urgency: :low
+  :resource_boundary: :cpu
+  :weight: 3
+  :idempotent:
+  :tags: []
 - :name: pipeline_default:ci_drop_pipeline
   :worker_name: Ci::DropPipelineWorker
   :feature_category: :continuous_integration

app/workers/ci/create_downstream_pipeline_worker.rb

+# frozen_string_literal: true
+
+module Ci
+  class CreateDownstreamPipelineWorker # rubocop:disable Scalability/IdempotentWorker
+    include ::ApplicationWorker
+    include ::PipelineQueue
+
+    sidekiq_options retry: 3
+    worker_resource_boundary :cpu
+
+    def perform(bridge_id)
+      ::Ci::Bridge.find_by_id(bridge_id).try do |bridge|
+        ::Ci::CreateDownstreamPipelineService
+          .new(bridge.project, bridge.user)
+          .execute(bridge)
+      end
+    end
+  end
+end

config/feature_flags/development/create_cross_project_pipeline_worker_rename.yml

+---
+name: create_cross_project_pipeline_worker_rename
+introduced_by_url: https://gitlab.com/gitlab-org/gitlab/-/merge_requests/70816
+rollout_issue_url: https://gitlab.com/gitlab-org/gitlab/-/issues/341410
+milestone: '14.4'
+type: development
+group: group::pipeline authoring
+default_enabled: false

ee/spec/models/ci/bridge_spec.rb

@@ -41,10 +41,28 @@ RSpec.describe Ci::Bridge do
         bridge.enqueue!
       end
 
-      it 'does not schedule downstream pipeline creation' do
-        bridge.enqueue!
+      context 'when the create_cross_project_pipeline_worker_rename feature is enabled' do
+        before do
+          stub_feature_flags(create_cross_project_pipeline_worker_rename: true)
+        end
+
+        it 'does not schedule downstream pipeline creation' do
+          bridge.enqueue!
+
+          expect(::Ci::CreateDownstreamPipelineWorker.jobs).to be_empty
+        end
+      end
+
+      context 'when the create_cross_project_pipeline_worker_rename feature is not enabled' do
+        before do
+          stub_feature_flags(create_cross_project_pipeline_worker_rename: false)
+        end
 
-        expect(::Ci::CreateCrossProjectPipelineWorker.jobs).to be_empty
+        it 'does not schedule downstream pipeline creation' do
+          bridge.enqueue!
+
+          expect(::Ci::CreateCrossProjectPipelineWorker.jobs).to be_empty
+        end
       end
     end
   end

spec/models/ci/bridge_spec.rb

@@ -73,21 +73,61 @@ RSpec.describe Ci::Bridge do
   describe 'state machine transitions' do
     context 'when bridge points towards downstream' do
       %i[created manual].each do |status|
-        it "schedules downstream pipeline creation when the status is #{status}" do
-          bridge.status = status
+        context 'when the create_cross_project_pipeline_worker_rename feature is enabled' do
+          before do
+            stub_feature_flags(create_cross_project_pipeline_worker_rename: true)
+          end
 
-          bridge.enqueue!
+          it "schedules downstream pipeline creation when the status is #{status}" do
+            bridge.status = status
 
-          expect(::Ci::CreateCrossProjectPipelineWorker.jobs.last['args']).to eq([bridge.id])
+            bridge.enqueue!
+
+            expect(::Ci::CreateDownstreamPipelineWorker.jobs.last['args']).to eq([bridge.id])
+          end
+        end
+
+        context 'when the create_cross_project_pipeline_worker_rename feature is not enabled' do
+          before do
+            stub_feature_flags(create_cross_project_pipeline_worker_rename: false)
+          end
+
+          it "schedules downstream pipeline creation when the status is #{status}" do
+            bridge.status = status
+
+            bridge.enqueue!
+
+            expect(::Ci::CreateCrossProjectPipelineWorker.jobs.last['args']).to eq([bridge.id])
+          end
         end
       end
 
-      it "schedules downstream pipeline creation when the status is waiting for resource" do
-        bridge.status = :waiting_for_resource
+      context 'when the create_cross_project_pipeline_worker_rename feature is enabled' do
+        before do
+          stub_feature_flags(create_cross_project_pipeline_worker_rename: true)
+        end
+
+        it "schedules downstream pipeline creation when the status is waiting for resource" do
+          bridge.status = :waiting_for_resource
+
+          bridge.enqueue_waiting_for_resource!
 
-        bridge.enqueue_waiting_for_resource!
+          expect(::Ci::CreateDownstreamPipelineWorker.jobs.last['args']).to eq([bridge.id])
+        end
+      end
 
-        expect(::Ci::CreateCrossProjectPipelineWorker.jobs.last['args']).to eq([bridge.id])
+      context 'when the create_cross_project_pipeline_worker_rename feature is not enabled' do
+        before do
+          stub_feature_flags(create_cross_project_pipeline_worker_rename: false)
+        end
+
+        it "schedules downstream pipeline creation when the status is waiting for resource" do
+          bridge.status = :waiting_for_resource
+
+          bridge.enqueue_waiting_for_resource!
+
+          expect(::Ci::CreateCrossProjectPipelineWorker.jobs.last['args']).to eq([bridge.id])
+        end
       end
 
       it 'raises error when the status is failed' do

spec/services/ci/play_bridge_service_spec.rb

@@ -23,18 +23,36 @@ RSpec.describe Ci::PlayBridgeService, '#execute' do
       expect(bridge.reload).to be_pending
     end
 
-    it 'enqueues Ci::CreateCrossProjectPipelineWorker' do
-      expect(::Ci::CreateCrossProjectPipelineWorker).to receive(:perform_async).with(bridge.id)
-
-      execute_service
-    end
-
     it "updates bridge's user" do
       execute_service
 
       expect(bridge.reload.user).to eq(user)
     end
 
+    context 'when the create_cross_project_pipeline_worker_rename feature is enabled' do
+      before do
+        stub_feature_flags(create_cross_project_pipeline_worker_rename: true)
+      end
+
+      it 'enqueues Ci::CreateDownstreamPipelineWorker' do
+        expect(::Ci::CreateDownstreamPipelineWorker).to receive(:perform_async).with(bridge.id)
+
+        execute_service
+      end
+    end
+
+    context 'when the create_cross_project_pipeline_worker_rename feature is not enabled' do
+      before do
+        stub_feature_flags(create_cross_project_pipeline_worker_rename: false)
+      end
+
+      it 'enqueues Ci::CreateCrossProjectPipelineWorker' do
+        expect(::Ci::CreateCrossProjectPipelineWorker).to receive(:perform_async).with(bridge.id)
+
+        execute_service
+      end
+    end
+
     context 'when a subsequent job is skipped' do
       let!(:job) { create(:ci_build, :skipped, pipeline: pipeline, stage_idx: bridge.stage_idx + 1) }
 

spec/workers/ci/create_downstream_pipeline_worker_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Ci::CreateDownstreamPipelineWorker do
+  let_it_be(:user) { create(:user) }
+  let_it_be(:project) { create(:project) }
+  let_it_be(:pipeline) { create(:ci_pipeline, project: project) }
+
+  let(:bridge) { create(:ci_bridge, user: user, pipeline: pipeline) }
+
+  let(:service) { double('pipeline creation service') }
+
+  describe '#perform' do
+    context 'when bridge exists' do
+      it 'calls cross project pipeline creation service' do
+        expect(Ci::CreateDownstreamPipelineService)
+          .to receive(:new)
+          .with(project, user)
+          .and_return(service)
+
+        expect(service).to receive(:execute).with(bridge)
+
+        described_class.new.perform(bridge.id)
+      end
+    end
+
+    context 'when bridge does not exist' do
+      it 'does nothing' do
+        expect(Ci::CreateDownstreamPipelineService)
+          .not_to receive(:new)
+
+        described_class.new.perform(non_existing_record_id)
+      end
+    end
+  end
+end

spec/workers/every_sidekiq_worker_spec.rb

@@ -155,6 +155,7 @@ RSpec.describe 'Every Sidekiq worker' do
         'Ci::BuildScheduleWorker' => 3,
         'Ci::BuildTraceChunkFlushWorker' => 3,
         'Ci::CreateCrossProjectPipelineWorker' => 3,
+        'Ci::CreateDownstreamPipelineWorker' => 3,
         'Ci::DailyBuildGroupReportResultsWorker' => 3,
         'Ci::DeleteObjectsWorker' => 0,
         'Ci::DropPipelineWorker' => 3,