Use AuthorizedProjectUpdate::ProjectCreateWorker for project create

Specialized workers use much less resource to update project
authorizations but it could lead to synchronization issues because it
doesn't serialize simultaneous update events. The previous approach is
not bulletproof either, inconsistencies do accumulate in the project
authorization cache over time. For now, we also use the previous
approach as a safety measure but with some delay and lower priority.

app/models/group.rb

@@ -305,9 +305,10 @@ class Group < Namespace
   # rubocop: enable CodeReuse/ServiceClass
 
   # rubocop: disable CodeReuse/ServiceClass
-  def refresh_members_authorized_projects(blocking: true)
-    UserProjectAccessChangedService.new(user_ids_for_project_authorizations)
-      .execute(blocking: blocking)
+  def refresh_members_authorized_projects(blocking: true, priority: UserProjectAccessChangedService::HIGH_PRIORITY)
+    UserProjectAccessChangedService
+      .new(user_ids_for_project_authorizations)
+      .execute(blocking: blocking, priority: priority)
   end
   # rubocop: enable CodeReuse/ServiceClass
 

app/services/clusters/management/create_project_service.rb

@@ -15,11 +15,8 @@ module Clusters
       def execute
         return unless management_project_required?
 
-        ActiveRecord::Base.transaction do
-          project = create_management_project!
-
-          update_cluster!(project)
-        end
+        project = create_management_project!
+        update_cluster!(project)
       end
 
       private

app/services/projects/create_service.rb

@@ -108,8 +108,22 @@ module Projects
     # users in the background
     def setup_authorizations
       if @project.group
-        @project.group.refresh_members_authorized_projects(blocking: false)
         current_user.refresh_authorized_projects
+
+        if Feature.enabled?(:specialized_project_authorization_workers)
+          AuthorizedProjectUpdate::ProjectCreateWorker.perform_async(@project.id)
+          # AuthorizedProjectsWorker uses an exclusive lease per user but
+          # specialized workers might have synchronization issues. Until we
+          # compare the inconsistency rates of both approaches, we still run
+          # AuthorizedProjectsWorker but with some delay and lower urgency as a
+          # safety net.
+          @project.group.refresh_members_authorized_projects(
+            blocking: false,
+            priority: UserProjectAccessChangedService::LOW_PRIORITY
+          )
+        else
+          @project.group.refresh_members_authorized_projects(blocking: false)
+        end
       else
         @project.add_maintainer(@project.namespace.owner, current_user: current_user)
       end

app/services/user_project_access_changed_service.rb

 # frozen_string_literal: true
 
 class UserProjectAccessChangedService
+  DELAY = 1.hour
+
+  HIGH_PRIORITY = :high
+  LOW_PRIORITY = :low
+
   def initialize(user_ids)
     @user_ids = Array.wrap(user_ids)
   end
 
-  def execute(blocking: true)
+  def execute(blocking: true, priority: HIGH_PRIORITY)
     bulk_args = @user_ids.map { |id| [id] }
 
     if blocking
       AuthorizedProjectsWorker.bulk_perform_and_wait(bulk_args)
     else
-      AuthorizedProjectsWorker.bulk_perform_async(bulk_args) # rubocop:disable Scalability/BulkPerformWithContext
+      if priority == HIGH_PRIORITY
+        AuthorizedProjectsWorker.bulk_perform_async(bulk_args) # rubocop:disable Scalability/BulkPerformWithContext
+      else
+        AuthorizedProjectUpdate::UserRefreshWithLowUrgencyWorker.bulk_perform_in(DELAY, bulk_args) # rubocop:disable Scalability/BulkPerformWithContext
+      end
     end
   end
 end

ee/app/services/ee/user_project_access_changed_service.rb

@@ -2,7 +2,7 @@
 
 module EE
   module UserProjectAccessChangedService
-    def execute(blocking: true)
+    def execute(blocking: true, priority: ::UserProjectAccessChangedService::HIGH_PRIORITY)
       result = super
 
       @user_ids.each do |id| # rubocop:disable Gitlab/ModuleWithInstanceVariables

spec/services/projects/create_service_spec.rb

@@ -510,6 +510,83 @@ describe Projects::CreateService, '#execute' do
     end
   end
 
+  context 'with specialized_project_authorization_workers' do
+    let_it_be(:other_user) { create(:user) }
+    let_it_be(:group) { create(:group) }
+
+    let(:opts) do
+      {
+        name: 'GitLab',
+        namespace_id: group.id
+      }
+    end
+
+    before do
+      group.add_maintainer(user)
+      group.add_developer(other_user)
+    end
+
+    it 'updates authorization for current_user' do
+      expect(Users::RefreshAuthorizedProjectsService).to(
+        receive(:new).with(user).and_call_original
+      )
+
+      project = create_project(user, opts)
+
+      expect(
+        Ability.allowed?(user, :read_project, project)
+      ).to be_truthy
+    end
+
+    it 'schedules authorization update for users with access to group' do
+      expect(AuthorizedProjectsWorker).not_to(
+        receive(:bulk_perform_async)
+      )
+      expect(AuthorizedProjectUpdate::ProjectCreateWorker).to(
+        receive(:perform_async).and_call_original
+      )
+      expect(AuthorizedProjectUpdate::UserRefreshWithLowUrgencyWorker).to(
+        receive(:bulk_perform_in)
+          .with(1.hour, array_including([user.id], [other_user.id]))
+          .and_call_original
+      )
+
+      create_project(user, opts)
+    end
+
+    context 'when feature is disabled' do
+      before do
+        stub_feature_flags(specialized_project_authorization_workers: false)
+      end
+
+      it 'updates authorization for current_user' do
+        expect(Users::RefreshAuthorizedProjectsService).to(
+          receive(:new).with(user).and_call_original
+        )
+
+        project = create_project(user, opts)
+
+        expect(
+          Ability.allowed?(user, :read_project, project)
+        ).to be_truthy
+      end
+
+      it 'uses AuthorizedProjectsWorker' do
+        expect(AuthorizedProjectsWorker).to(
+          receive(:bulk_perform_async).with(array_including([user.id], [other_user.id])).and_call_original
+        )
+        expect(AuthorizedProjectUpdate::ProjectCreateWorker).not_to(
+          receive(:perform_async)
+        )
+        expect(AuthorizedProjectUpdate::UserRefreshWithLowUrgencyWorker).not_to(
+          receive(:bulk_perform_in)
+        )
+
+        create_project(user, opts)
+      end
+    end
+  end
+
   def create_project(user, opts)
     Projects::CreateService.new(user, opts).execute
   end

spec/services/user_project_access_changed_service_spec.rb

@@ -17,5 +17,14 @@ describe UserProjectAccessChangedService do
 
       described_class.new([1, 2]).execute(blocking: false)
     end
+
+    it 'permits low-priority operation' do
+      expect(AuthorizedProjectUpdate::UserRefreshWithLowUrgencyWorker).to(
+        receive(:bulk_perform_in).with(described_class::DELAY, [[1], [2]])
+      )
+
+      described_class.new([1, 2]).execute(blocking: false,
+                                          priority: described_class::LOW_PRIORITY)
+    end
   end
 end