Use specialized worker for project share

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/services/projects/group_links/create_service.rb

@@ -13,12 +13,32 @@ module Projects
         )
 
         if link.save
-          group.refresh_members_authorized_projects
+          setup_authorizations(group)
           success(link: link)
         else
           error(link.errors.full_messages.to_sentence, 409)
         end
       end
+
+      private
+
+      def setup_authorizations(group)
+        if Feature.enabled?(:specialized_project_authorization_project_share_worker)
+          AuthorizedProjectUpdate::ProjectGroupLinkCreateWorker.perform_async(project.id, group.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.
+          group.refresh_members_authorized_projects(
+            blocking: false,
+            priority: UserProjectAccessChangedService::LOW_PRIORITY
+          )
+        else
+          group.refresh_members_authorized_projects(blocking: false)
+        end
+      end
     end
   end
 end

spec/requests/api/projects_spec.rb

@@ -2133,7 +2133,7 @@ describe API::Projects do
       expect(json_response['expires_at']).to eq(expires_at.to_s)
     end
 
-    it 'updates project authorization' do
+    it 'updates project authorization', :sidekiq_inline do
       expect do
         post api("/projects/#{project.id}/share", user), params: { group_id: group.id, group_access: Gitlab::Access::DEVELOPER }
       end.to(

spec/services/projects/group_links/create_service_spec.rb

@@ -23,7 +23,7 @@ describe Projects::GroupLinks::CreateService, '#execute' do
     expect { subject.execute(group) }.to change { project.project_group_links.count }.from(0).to(1)
   end
 
-  it 'updates authorization' do
+  it 'updates authorization', :sidekiq_inline do
     expect { subject.execute(group) }.to(
       change { Ability.allowed?(user, :read_project, project) }
         .from(false).to(true))
@@ -36,4 +36,50 @@ describe Projects::GroupLinks::CreateService, '#execute' do
   it 'returns error if user is not allowed to share with a group' do
     expect { subject.execute(create(:group)) }.not_to change { project.project_group_links.count }
   end
+
+  context 'with specialized_project_authorization_workers' do
+    let_it_be(:other_user) { create(:user) }
+
+    before do
+      group.add_developer(other_user)
+    end
+
+    it 'schedules authorization update for users with access to group' do
+      expect(AuthorizedProjectsWorker).not_to(
+        receive(:bulk_perform_async)
+      )
+      expect(AuthorizedProjectUpdate::ProjectGroupLinkCreateWorker).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]),
+                batch_delay: 30.seconds, batch_size: 100)
+          .and_call_original
+      )
+
+      subject.execute(group)
+    end
+
+    context 'when feature is disabled' do
+      before do
+        stub_feature_flags(specialized_project_authorization_project_share_worker: false)
+      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)
+        )
+
+        subject.execute(group)
+      end
+    end
+  end
 end