Move Minutes Usage Update to Async Job

app/services/ci/destroy_pipeline_service.rb

@@ -7,7 +7,7 @@ module Ci
 
       Ci::ExpirePipelineCacheService.new.execute(pipeline, delete: true)
 
-      pipeline.cancel_running if pipeline.cancelable? && ::Feature.enabled?(:cancel_pipelines_prior_to_destroy, default_enabled: :yaml)
+      pipeline.cancel_running if pipeline.cancelable? && ::Feature.enabled?(:cancel_pipelines_prior_to_destroy, pipeline.project, default_enabled: :yaml)
 
       pipeline.reset.destroy!
 

ee/app/models/ci/minutes/namespace_monthly_usage.rb

@@ -22,8 +22,8 @@ module Ci
       #
       # Here we will also do any recalculation of additional minutes based on the
       # previous month usage.
-      def self.find_or_create_current(namespace)
-        current_month.safe_find_or_create_by(namespace: namespace)
+      def self.find_or_create_current(namespace_id:)
+        current_month.safe_find_or_create_by(namespace_id: namespace_id)
       end
 
       def self.increase_usage(usage, amount)

ee/app/models/ci/minutes/project_monthly_usage.rb

@@ -19,8 +19,8 @@ module Ci
       # since this will lazily create an entry if it doesn't exist.
       # For example, on the 1st of each month, when we update the usage for a project,
       # we will automatically generate new records and reset usage for the current month.
-      def self.find_or_create_current(project)
-        current_month.safe_find_or_create_by(project: project)
+      def self.find_or_create_current(project_id:)
+        current_month.safe_find_or_create_by(project_id: project_id)
       end
 
       def self.increase_usage(usage, amount)

ee/app/models/ee/ci/build.rb

@@ -52,9 +52,7 @@ module EE
         state_machine :status do
           after_transition any => [:success, :failed, :canceled] do |build|
             build.run_after_commit do
-              # TODO(Issue #331891): before enabling this feature flag. Move update consumption to async while keeping consumption calculation sync.
-              # This will ensure consumption is calculated before related records are deleted.
-              if ::Feature.enabled?(:cancel_pipelines_prior_to_destroy, default_enabled: :yaml)
+              if ::Feature.enabled?(:cancel_pipelines_prior_to_destroy, build.project, default_enabled: :yaml)
                 ::Ci::Minutes::UpdateBuildMinutesService.new(build.project, nil).execute(build)
               end
             end

ee/app/services/ci/minutes/update_build_minutes_service.rb

@@ -14,14 +14,20 @@ module Ci
 
         return unless consumption > 0
 
-        # TODO(Issue #335338): Introduce async worker UpdateProjectAndNamespaceUsageWorker
-        Ci::Minutes::UpdateProjectAndNamespaceUsageService.new(project, namespace).execute(consumption)
-
+        update_minutes(consumption)
         compare_with_live_consumption(build, consumption)
       end
 
       private
 
+      def update_minutes(consumption)
+        if ::Feature.enabled?(:cancel_pipelines_prior_to_destroy, project, default_enabled: :yaml)
+          ::Ci::Minutes::UpdateProjectAndNamespaceUsageWorker.perform_async(consumption, project.id, namespace.id)
+        else
+          ::Ci::Minutes::UpdateProjectAndNamespaceUsageService.new(project.id, namespace.id).execute(consumption)
+        end
+      end
+
       def compare_with_live_consumption(build, consumption)
         live_consumption = ::Ci::Minutes::TrackLiveConsumptionService.new(build).live_consumption
         return if live_consumption == 0

ee/app/services/ci/minutes/update_project_and_namespace_usage_service.rb

@@ -3,53 +3,83 @@
 module Ci
   module Minutes
     class UpdateProjectAndNamespaceUsageService
-      def initialize(project, namespace)
-        @project = project
-        @namespace = namespace
+      include Gitlab::Utils::StrongMemoize
+
+      def initialize(project_id, namespace_id)
+        @project_id = project_id
+        @namespace_id = namespace_id
+        # TODO(issue 335885): Use project_id only and don't query for projects which may be deleted
+        @project = Project.find_by_id(project_id)
       end
 
       # Updates the project and namespace usage based on the passed consumption amount
       def execute(consumption)
-        consumption_in_seconds = consumption.minutes.to_i
-        legacy_track_usage_of_monthly_minutes(consumption_in_seconds)
-
+        legacy_track_usage_of_monthly_minutes(consumption)
+        ApplicationRecord.transaction do
           track_usage_of_monthly_minutes(consumption)
+
           send_minutes_email_notification
         end
+      end
 
       private
 
       def send_minutes_email_notification
         # `perform reset` on `project` because `Namespace#namespace_statistics` will otherwise return stale data.
+        # TODO(issue 335885): Remove @project
         ::Ci::Minutes::EmailNotificationService.new(@project.reset).execute if ::Gitlab.com?
       end
 
-      def legacy_track_usage_of_monthly_minutes(consumption_in_seconds)
-        ProjectStatistics.update_counters(project_statistics,
-          shared_runners_seconds: consumption_in_seconds)
+      def legacy_track_usage_of_monthly_minutes(consumption)
+        consumption_in_seconds = consumption.minutes.to_i
 
-        NamespaceStatistics.update_counters(namespace_statistics,
-          shared_runners_seconds: consumption_in_seconds)
+        update_legacy_project_minutes(consumption_in_seconds)
+        update_legacy_namespace_minutes(consumption_in_seconds)
       end
 
       def track_usage_of_monthly_minutes(consumption)
+        # TODO(issue 335885): Remove @project
         return unless Feature.enabled?(:ci_minutes_monthly_tracking, @project, default_enabled: :yaml)
 
-        namespace_usage = ::Ci::Minutes::NamespaceMonthlyUsage.find_or_create_current(@namespace)
-        project_usage = ::Ci::Minutes::ProjectMonthlyUsage.find_or_create_current(@project)
+        ::Ci::Minutes::NamespaceMonthlyUsage.increase_usage(namespace_usage, consumption) if namespace_usage
+        ::Ci::Minutes::ProjectMonthlyUsage.increase_usage(project_usage, consumption) if project_usage
+      end
 
-        ApplicationRecord.transaction do
-          ::Ci::Minutes::NamespaceMonthlyUsage.increase_usage(namespace_usage, consumption)
-          ::Ci::Minutes::ProjectMonthlyUsage.increase_usage(project_usage, consumption)
+      def update_legacy_project_minutes(consumption_in_seconds)
+        if project_statistics
+          ProjectStatistics.update_counters(project_statistics, shared_runners_seconds: consumption_in_seconds)
+        end
+      end
+
+      def update_legacy_namespace_minutes(consumption_in_seconds)
+        if namespace_statistics
+          NamespaceStatistics.update_counters(namespace_statistics, shared_runners_seconds: consumption_in_seconds)
+        end
+      end
+
+      def namespace_usage
+        ::Ci::Minutes::NamespaceMonthlyUsage.find_or_create_current(namespace_id: @namespace_id)
+      end
+
+      def project_usage
+        strong_memoize(:project_usage) do
+          ::Ci::Minutes::ProjectMonthlyUsage.find_or_create_current(project_id: @project_id)
+        rescue ActiveRecord::InvalidForeignKey
         end
       end
 
       def namespace_statistics
-        @namespace.namespace_statistics || @namespace.create_namespace_statistics
+        strong_memoize(:namespace_statistics) do
+          NamespaceStatistics.safe_find_or_create_by!(namespace_id: @namespace_id)
+        rescue ActiveRecord::NotNullViolation, ActiveRecord::RecordInvalid
+        end
       end
 
       def project_statistics
-        @project.statistics || @project.create_statistics(namespace: @project.namespace)
+        strong_memoize(:project_statistics) do
+          ProjectStatistics.safe_find_or_create_by!(project_id: @project_id)
+        rescue ActiveRecord::NotNullViolation, ActiveRecord::RecordInvalid
+        end
       end
     end
   end

ee/app/workers/all_queues.yml

@@ -791,6 +791,15 @@
   :weight: 1
   :idempotent:
   :tags: []
+- :name: pipeline_background:ci_minutes_update_project_and_namespace_usage
+  :worker_name: Ci::Minutes::UpdateProjectAndNamespaceUsageWorker
+  :feature_category: :continuous_integration
+  :has_external_dependencies:
+  :urgency: :low
+  :resource_boundary: :unknown
+  :weight: 1
+  :idempotent:
+  :tags: []
 - :name: pipeline_background:ci_sync_reports_to_report_approval_rules
   :worker_name: Ci::SyncReportsToReportApprovalRulesWorker
   :feature_category: :continuous_integration

ee/app/workers/ci/minutes/update_project_and_namespace_usage_worker.rb

+# frozen_string_literal: true
+
+module Ci
+  module Minutes
+    class UpdateProjectAndNamespaceUsageWorker # rubocop:disable Scalability/IdempotentWorker
+      include ApplicationWorker
+      include PipelineBackgroundQueue
+
+      urgency :low
+      data_consistency :always # primarily performs writes
+
+      def perform(consumption, project_id, namespace_id)
+        ::Ci::Minutes::UpdateProjectAndNamespaceUsageService.new(project_id, namespace_id).execute(consumption)
+      end
+    end
+  end
+end

ee/app/workers/ee/ci/build_finished_worker.rb

@@ -4,7 +4,7 @@ module EE
   module Ci
     module BuildFinishedWorker
       def process_build(build)
-        unless ::Feature.enabled?(:cancel_pipelines_prior_to_destroy, default_enabled: :yaml)
+        unless ::Feature.enabled?(:cancel_pipelines_prior_to_destroy, build.project, default_enabled: :yaml)
           ::Ci::Minutes::UpdateBuildMinutesService.new(build.project, nil).execute(build)
         end
 

ee/spec/models/ci/minutes/namespace_monthly_usage_spec.rb

@@ -22,7 +22,7 @@ RSpec.describe Ci::Minutes::NamespaceMonthlyUsage do
   end
 
   describe '.find_or_create_current' do
-    subject { described_class.find_or_create_current(namespace) }
+    subject { described_class.find_or_create_current(namespace_id: namespace.id) }
 
     shared_examples 'creates usage record' do
       it 'creates new record and resets minutes consumption' do

ee/spec/models/ci/minutes/project_monthly_usage_spec.rb

@@ -22,7 +22,7 @@ RSpec.describe Ci::Minutes::ProjectMonthlyUsage do
   end
 
   describe '.find_or_create_current' do
-    subject { described_class.find_or_create_current(project) }
+    subject { described_class.find_or_create_current(project_id: project.id) }
 
     shared_examples 'creates usage record' do
       it 'creates new record and resets minutes consumption' do

ee/spec/services/ci/minutes/update_build_minutes_service_spec.rb

@@ -3,7 +3,6 @@
 require 'spec_helper'
 
 RSpec.describe Ci::Minutes::UpdateBuildMinutesService do
-  describe '#perform' do
   let(:namespace) { create(:namespace, shared_runners_minutes_limit: 100) }
   let(:project) { create(:project, :private, namespace: namespace) }
   let(:pipeline) { create(:ci_pipeline, project: project) }
@@ -14,17 +13,18 @@ RSpec.describe Ci::Minutes::UpdateBuildMinutesService do
       started_at: 2.hours.ago, finished_at: 1.hour.ago)
   end
 
-    let(:namespace_amount_used) { Ci::Minutes::NamespaceMonthlyUsage.find_or_create_current(namespace).amount_used }
-    let(:project_amount_used) { Ci::Minutes::ProjectMonthlyUsage.find_or_create_current(project).amount_used }
+  let(:namespace_amount_used) { Ci::Minutes::NamespaceMonthlyUsage.find_or_create_current(namespace_id: namespace.id).amount_used }
+  let(:project_amount_used) { Ci::Minutes::ProjectMonthlyUsage.find_or_create_current(project_id: project.id).amount_used }
 
   subject { described_class.new(project, nil).execute(build) }
 
+  shared_examples 'executes service' do
     shared_examples 'new tracking matches legacy tracking' do
       it 'stores the same information in both legacy and new tracking' do
         subject
 
         expect(namespace_amount_used)
-          .to eq((namespace.namespace_statistics.reload.shared_runners_seconds.to_f / 60).round(2))
+          .to eq((namespace.reload.namespace_statistics.shared_runners_seconds.to_f / 60).round(2))
 
         expect(project_amount_used)
           .to eq((project.statistics.reload.shared_runners_seconds.to_f / 60).round(2))
@@ -181,7 +181,7 @@ RSpec.describe Ci::Minutes::UpdateBuildMinutesService do
         let(:root_ancestor) { create(:group, shared_runners_minutes_limit: 100) }
         let(:namespace) { create(:group, parent: root_ancestor) }
 
-        let(:namespace_amount_used) { Ci::Minutes::NamespaceMonthlyUsage.find_or_create_current(root_ancestor).amount_used }
+        let(:namespace_amount_used) { Ci::Minutes::NamespaceMonthlyUsage.find_or_create_current(namespace_id: root_ancestor.id).amount_used }
 
         it 'creates a statistics in root group' do
           subject
@@ -253,4 +253,18 @@ RSpec.describe Ci::Minutes::UpdateBuildMinutesService do
       it_behaves_like 'does nothing'
     end
   end
+
+  describe '#execute' do
+    context 'when cancel_pipelines_prior_to_destroy enabled', :sidekiq_inline do
+      include_examples 'executes service'
+    end
+
+    context 'when cancel_pipelines_prior_to_destroy disabled' do
+      before do
+        stub_feature_flags(cancel_pipelines_prior_to_destroy: false)
+      end
+
+      include_examples 'executes service'
+    end
+  end
 end

ee/spec/services/ci/minutes/update_project_and_namespace_usage_service_spec.rb

@@ -3,16 +3,16 @@
 require 'spec_helper'
 
 RSpec.describe Ci::Minutes::UpdateProjectAndNamespaceUsageService do
-  let_it_be(:namespace) { create(:namespace, shared_runners_minutes_limit: 100) }
-  let_it_be(:project) { create(:project, :private, namespace: namespace) }
+  let_it_be(:project) { create(:project, :private) }
+  let_it_be(:namespace) { project.namespace }
 
   let(:consumption_minutes) { 120 }
   let(:consumption_seconds) { consumption_minutes * 60 }
-  let(:namespace_amount_used) { Ci::Minutes::NamespaceMonthlyUsage.find_or_create_current(namespace).amount_used }
-  let(:project_amount_used) { Ci::Minutes::ProjectMonthlyUsage.find_or_create_current(project).amount_used }
+  let(:namespace_amount_used) { Ci::Minutes::NamespaceMonthlyUsage.find_or_create_current(namespace_id: namespace.id).amount_used }
+  let(:project_amount_used) { Ci::Minutes::ProjectMonthlyUsage.find_or_create_current(project_id: project.id).amount_used }
 
   describe '#execute' do
-    subject { described_class.new(project, namespace).execute(consumption_minutes) }
+    subject { described_class.new(project.id, namespace.id).execute(consumption_minutes) }
 
     context 'with shared runner' do
       context 'when statistics and usage do not have existing values' do
@@ -26,6 +26,46 @@ RSpec.describe Ci::Minutes::UpdateProjectAndNamespaceUsageService do
             .to eq(consumption_seconds)
         end
 
+        context 'when project deleted' do
+          let(:project) { double(id: non_existing_record_id) }
+
+          it 'will complete successfully and increment namespace statistics' do
+            subject
+
+            expect(ProjectStatistics.find_by_project_id(project.id)).to be_nil
+            expect(NamespaceStatistics.find_by_namespace_id(namespace.id).shared_runners_seconds).to eq(consumption_seconds)
+            expect(Ci::Minutes::ProjectMonthlyUsage.find_by_project_id(project.id)).to be_nil
+            expect(Ci::Minutes::NamespaceMonthlyUsage.find_by_namespace_id(namespace.id).amount_used).to eq(consumption_minutes)
+          end
+        end
+
+        context 'when namespace deleted' do
+          let(:namespace) { double(id: non_existing_record_id) }
+
+          it 'will complete successfully' do
+            subject
+
+            expect(ProjectStatistics.find_by_project_id(project.id).shared_runners_seconds).to eq(consumption_seconds)
+            expect(NamespaceStatistics.find_by_namespace_id(namespace.id)).to be_nil
+            expect(Ci::Minutes::ProjectMonthlyUsage.find_by_project_id(project.id).amount_used).to eq(consumption_minutes)
+            expect(Ci::Minutes::NamespaceMonthlyUsage.find_by_namespace_id(namespace.id).amount_used).to eq(consumption_minutes)
+          end
+        end
+
+        context 'when project and namespace deleted' do
+          let(:project) { double(id: non_existing_record_id) }
+          let(:namespace) { double(id: non_existing_record_id) }
+
+          it 'will complete successfully' do
+            subject
+
+            expect(ProjectStatistics.find_by_project_id(project.id)).to be_nil
+            expect(NamespaceStatistics.find_by_namespace_id(namespace.id)).to be_nil
+            expect(Ci::Minutes::ProjectMonthlyUsage.find_by_project_id(project.id)).to be_nil
+            expect(Ci::Minutes::NamespaceMonthlyUsage.find_by_namespace_id(namespace.id).amount_used).to eq(consumption_minutes)
+          end
+        end
+
         it 'updates monthly usage with consumption minutes' do
           subject
 

ee/spec/workers/ci/minutes/update_project_and_namespace_usage_worker_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Ci::Minutes::UpdateProjectAndNamespaceUsageWorker do
+  let_it_be(:project) { create(:project) }
+  let_it_be(:namespace) { project.namespace }
+
+  let(:consumption) { 100 }
+  let(:consumption_seconds) { consumption * 60 }
+  let(:worker) { described_class.new }
+
+  describe '#perform' do
+    it 'executes UpdateProjectAndNamespaceUsageService' do
+      service_instance = double
+      expect(::Ci::Minutes::UpdateProjectAndNamespaceUsageService).to receive(:new).with(project.id, namespace.id).and_return(service_instance)
+      expect(service_instance).to receive(:execute).with(consumption)
+
+      worker.perform(consumption, project.id, namespace.id)
+    end
+
+    it 'updates statistics and usage' do
+      worker.perform(consumption, project.id, namespace.id)
+
+      expect(project.statistics.reload.shared_runners_seconds).to eq(consumption_seconds)
+      expect(namespace.namespace_statistics.reload.shared_runners_seconds).to eq(consumption_seconds)
+      expect(Ci::Minutes::NamespaceMonthlyUsage.find_by(namespace: namespace).amount_used).to eq(consumption)
+      expect(Ci::Minutes::ProjectMonthlyUsage.find_by(project: project).amount_used).to eq(consumption)
+    end
+
+    it 'accumulates only legacy statistics on failure (behaves transactionally)' do
+      allow(Ci::Minutes::ProjectMonthlyUsage).to receive(:new).and_raise(StandardError)
+
+      expect { worker.perform(consumption, project.id, namespace.id) }.to raise_error(StandardError)
+
+      expect(project.reload.statistics.shared_runners_seconds).to eq(consumption_seconds)
+      expect(namespace.reload.namespace_statistics.shared_runners_seconds).to eq(consumption_seconds)
+      expect(Ci::Minutes::NamespaceMonthlyUsage.find_by(namespace: namespace)).to eq(nil)
+      expect(Ci::Minutes::ProjectMonthlyUsage.find_by(project: project)).to eq(nil)
+      expect(::Ci::Minutes::EmailNotificationService).not_to receive(:new)
+    end
+  end
+end

lib/gitlab/ci/features.rb

@@ -3,7 +3,7 @@
 module Gitlab
   module Ci
     ##
-    # Ci::Features is a class that aggregates all CI/CD feature flags in one place.
+    # Deprecated: Ci::Features is a class that aggregates all CI/CD feature flags in one place.
     #
     module Features
       # NOTE: The feature flag `disallow_to_create_merge_request_pipelines_in_target_project`