Fix database cross-join in Project & Group transfer

app/services/ci/update_pending_build_service.rb

@@ -9,7 +9,7 @@ module Ci
 
     def initialize(model, update_params)
       @model = model
-      @update_params = update_params
+      @update_params = update_params.symbolize_keys
 
       validations!
     end

app/services/groups/transfer_service.rb

@@ -29,11 +29,11 @@ module Groups
         update_group_attributes
         ensure_ownership
         update_integrations
-        update_pending_builds!
       end
 
       post_update_hooks(@updated_project_ids)
       propagate_integrations
+      update_pending_builds
 
       true
     end
@@ -228,13 +228,19 @@ module Groups
       end
     end
 
-    def update_pending_builds!
-      update_params = {
+    def update_pending_builds
+      if Feature.enabled?(:ci_pending_builds_async_update, default_enabled: :yaml)
+        ::Ci::PendingBuilds::UpdateGroupWorker.perform_async(group.id, pending_builds_params)
+      else
+        ::Ci::UpdatePendingBuildService.new(group, pending_builds_params).execute
+      end
+    end
+
+    def pending_builds_params
+      {
         namespace_traversal_ids: group.traversal_ids,
         namespace_id: group.id
       }
-
-      ::Ci::UpdatePendingBuildService.new(group, update_params).execute
     end
   end
 end

app/services/projects/transfer_service.rb

@@ -104,10 +104,10 @@ module Projects
         update_repository_configuration(@new_path)
 
         execute_system_hooks
-
-        update_pending_builds!
       end
 
+      update_pending_builds
+
       post_update_hooks(project)
     rescue Exception # rubocop:disable Lint/RescueException
       rollback_side_effects
@@ -244,13 +244,19 @@ module Projects
       Integration.create_from_active_default_integrations(project, :project_id)
     end
 
-    def update_pending_builds!
-      update_params = {
+    def update_pending_builds
+      if Feature.enabled?(:ci_pending_builds_async_update, default_enabled: :yaml)
+        ::Ci::PendingBuilds::UpdateProjectWorker.perform_async(project.id, pending_builds_params)
+      else
+        ::Ci::UpdatePendingBuildService.new(project, pending_builds_params).execute
+      end
+    end
+
+    def pending_builds_params
+      {
         namespace_id: new_namespace.id,
         namespace_traversal_ids: new_namespace.traversal_ids
       }
-
-      ::Ci::UpdatePendingBuildService.new(project, update_params).execute
     end
   end
 end

app/workers/all_queues.yml

@@ -1447,6 +1447,24 @@
   :weight: 1
   :idempotent: true
   :tags: []
+- :name: pipeline_background:ci_pending_builds_update_group
+  :worker_name: Ci::PendingBuilds::UpdateGroupWorker
+  :feature_category: :continuous_integration
+  :has_external_dependencies:
+  :urgency: :low
+  :resource_boundary: :unknown
+  :weight: 1
+  :idempotent: true
+  :tags: []
+- :name: pipeline_background:ci_pending_builds_update_project
+  :worker_name: Ci::PendingBuilds::UpdateProjectWorker
+  :feature_category: :continuous_integration
+  :has_external_dependencies:
+  :urgency: :low
+  :resource_boundary: :unknown
+  :weight: 1
+  :idempotent: true
+  :tags: []
 - :name: pipeline_background:ci_pipeline_artifacts_coverage_report
   :worker_name: Ci::PipelineArtifacts::CoverageReportWorker
   :feature_category: :code_testing

app/workers/ci/pending_builds/update_group_worker.rb

+# frozen_string_literal: true
+
+module Ci
+  module PendingBuilds
+    class UpdateGroupWorker
+      include ApplicationWorker
+      include PipelineBackgroundQueue
+
+      data_consistency :always
+      idempotent!
+
+      def perform(group_id, update_params)
+        ::Group.find_by_id(group_id).try do |group|
+          ::Ci::UpdatePendingBuildService.new(group, update_params).execute
+        end
+      end
+    end
+  end
+end

app/workers/ci/pending_builds/update_project_worker.rb

+# frozen_string_literal: true
+
+module Ci
+  module PendingBuilds
+    class UpdateProjectWorker
+      include ApplicationWorker
+      include PipelineBackgroundQueue
+
+      data_consistency :always
+      idempotent!
+
+      def perform(project_id, update_params)
+        ::Project.find_by_id(project_id).try do |project|
+          ::Ci::UpdatePendingBuildService.new(project, update_params).execute
+        end
+      end
+    end
+  end
+end

config/feature_flags/development/ci_pending_builds_async_update.yml

+---
+name: ci_pending_builds_async_update
+introduced_by_url: https://gitlab.com/gitlab-org/gitlab/-/merge_requests/75197
+rollout_issue_url: https://gitlab.com/gitlab-org/gitlab/-/issues/346641
+milestone: '14.6'
+type: development
+group: group::pipeline execution
+default_enabled: false

spec/services/groups/transfer_service_spec.rb

@@ -792,7 +792,7 @@ RSpec.describe Groups::TransferService, :sidekiq_inline do
         end
       end
 
-      context 'when group has pending builds' do
+      context 'when group has pending builds', :sidekiq_inline do
         let_it_be(:project) { create(:project, :public, namespace: group.reload) }
         let_it_be(:other_project) { create(:project) }
         let_it_be(:pending_build) { create(:ci_pending_build, project: project) }
@@ -814,6 +814,20 @@ RSpec.describe Groups::TransferService, :sidekiq_inline do
           expect(unrelated_pending_build.namespace_id).to eq(other_project.namespace_id)
           expect(unrelated_pending_build.namespace_traversal_ids).to eq(other_project.namespace.traversal_ids)
         end
+
+        context 'when ci_pending_builds_async_update is disabled' do
+          let(:update_pending_build_service) { instance_double(::Ci::PendingBuilds::UpdateGroupWorker) }
+
+          before do
+            stub_feature_flags(ci_pending_builds_async_update: false)
+          end
+
+          it 'does not call the new worker' do
+            expect(::Ci::PendingBuilds::UpdateGroupWorker).not_to receive(:perform_async)
+
+            transfer_service.execute(new_parent_group)
+          end
+        end
       end
     end
 

spec/services/projects/transfer_service_spec.rb

@@ -169,7 +169,7 @@ RSpec.describe Projects::TransferService do
       end
     end
 
-    context 'when project has pending builds' do
+    context 'when project has pending builds', :sidekiq_inline do
       let!(:other_project) { create(:project) }
       let!(:pending_build) { create(:ci_pending_build, project: project.reload) }
       let!(:unrelated_pending_build) { create(:ci_pending_build, project: other_project) }
@@ -189,6 +189,20 @@ RSpec.describe Projects::TransferService do
         expect(unrelated_pending_build.namespace_id).to eq(other_project.namespace_id)
         expect(unrelated_pending_build.namespace_traversal_ids).to eq(other_project.namespace.traversal_ids)
       end
+
+      context 'when ci_pending_builds_async_update is disabled' do
+        let(:update_pending_build_service) { instance_double(::Ci::PendingBuilds::UpdateProjectWorker) }
+
+        before do
+          stub_feature_flags(ci_pending_builds_async_update: false)
+        end
+
+        it 'does not call the new worker' do
+          expect(::Ci::PendingBuilds::UpdateProjectWorker).not_to receive(:perform_async)
+
+          execute_transfer
+        end
+      end
     end
   end
 
@@ -251,7 +265,7 @@ RSpec.describe Projects::TransferService do
       )
     end
 
-    context 'when project has pending builds' do
+    context 'when project has pending builds', :sidekiq_inline do
       let!(:other_project) { create(:project) }
       let!(:pending_build) { create(:ci_pending_build, project: project.reload) }
       let!(:unrelated_pending_build) { create(:ci_pending_build, project: other_project) }

spec/support/database/cross-database-modification-allowlist.yml

@@ -78,10 +78,8 @@
 - "./spec/services/ci/pipeline_bridge_status_service_spec.rb"
 - "./spec/services/ci/pipelines/add_job_service_spec.rb"
 - "./spec/services/ci/retry_build_service_spec.rb"
-- "./spec/services/groups/transfer_service_spec.rb"
 - "./spec/services/projects/destroy_service_spec.rb"
 - "./spec/services/projects/overwrite_project_service_spec.rb"
-- "./spec/services/projects/transfer_service_spec.rb"
 - "./spec/services/resource_access_tokens/revoke_service_spec.rb"
 - "./spec/services/users/destroy_service_spec.rb"
 - "./spec/services/users/reject_service_spec.rb"

spec/workers/ci/pending_builds/update_group_worker_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Ci::PendingBuilds::UpdateGroupWorker do
+  describe '#perform' do
+    let(:worker) { described_class.new }
+
+    context 'when a group is not provided' do
+      it 'does not call the service' do
+        expect(::Ci::UpdatePendingBuildService).not_to receive(:new)
+      end
+    end
+
+    context 'when everything is ok' do
+      let(:group) { create(:group) }
+      let(:update_pending_build_service) { instance_double(::Ci::UpdatePendingBuildService) }
+      let(:update_params) { { "namespace_id" => group.id } }
+
+      it 'calls the service' do
+        expect(::Ci::UpdatePendingBuildService).to receive(:new).with(group, update_params).and_return(update_pending_build_service)
+        expect(update_pending_build_service).to receive(:execute)
+
+        worker.perform(group.id, update_params)
+      end
+
+      include_examples 'an idempotent worker' do
+        let(:pending_build) { create(:ci_pending_build) }
+        let(:update_params) { { "namespace_id" => pending_build.namespace_id } }
+        let(:job_args) { [pending_build.namespace_id, update_params] }
+
+        it 'updates the pending builds' do
+          subject
+
+          expect(pending_build.reload.namespace_id).to eq(update_params["namespace_id"])
+        end
+      end
+    end
+  end
+end

spec/workers/ci/pending_builds/update_project_worker_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Ci::PendingBuilds::UpdateProjectWorker do
+  describe '#perform' do
+    let(:worker) { described_class.new }
+
+    context 'when a project is not provided' do
+      it 'does not call the service' do
+        expect(::Ci::UpdatePendingBuildService).not_to receive(:new)
+      end
+    end
+
+    context 'when everything is ok' do
+      let(:project) { create(:project) }
+      let(:group) { create(:group) }
+      let(:update_pending_build_service) { instance_double(::Ci::UpdatePendingBuildService) }
+      let(:update_params) { { "namespace_id" => group.id } }
+
+      it 'calls the service' do
+        expect(::Ci::UpdatePendingBuildService).to receive(:new).with(project, update_params).and_return(update_pending_build_service)
+        expect(update_pending_build_service).to receive(:execute)
+
+        worker.perform(project.id, update_params)
+      end
+
+      include_examples 'an idempotent worker' do
+        let(:pending_build) { create(:ci_pending_build) }
+        let(:job_args) { [pending_build.project_id, update_params] }
+
+        it 'updates the pending builds' do
+          subject
+
+          expect(pending_build.reload.namespace_id).to eq(update_params["namespace_id"])
+        end
+      end
+    end
+  end
+end