Expire job and pipeline cache synchronously

This avoids the delay we are currently experiencing with scheduled
Sidekiq jobs and saves us from Sidekiq overhead.

app/models/ci/pipeline.rb

@@ -236,7 +236,12 @@ module Ci
 
         pipeline.run_after_commit do
           PipelineHooksWorker.perform_async(pipeline.id)
-          ExpirePipelineCacheWorker.perform_async(pipeline.id)
+
+          if Feature.enabled?(:expire_job_and_pipeline_cache_synchronously, pipeline.project, default_enabled: :yaml)
+            Ci::ExpirePipelineCacheService.new.execute(pipeline) # rubocop: disable CodeReuse/ServiceClass
+          else
+            ExpirePipelineCacheWorker.perform_async(pipeline.id)
+          end
         end
       end
 

app/models/commit_status.rb

@@ -188,7 +188,12 @@ class CommitStatus < Ci::ApplicationRecord
 
       commit_status.run_after_commit do
         PipelineProcessWorker.perform_async(pipeline_id) unless transition_options[:skip_pipeline_processing]
-        ExpireJobCacheWorker.perform_async(id)
+
+        if Feature.enabled?(:expire_job_and_pipeline_cache_synchronously, project, default_enabled: :yaml)
+          expire_etag_cache!
+        else
+          ExpireJobCacheWorker.perform_async(id)
+        end
       end
     end
 
@@ -301,6 +306,12 @@ class CommitStatus < Ci::ApplicationRecord
       .update_all(retried: true, processed: true)
   end
 
+  def expire_etag_cache!
+    job_path = Gitlab::Routing.url_helpers.project_build_path(project, id, format: :json)
+
+    Gitlab::EtagCaching::Store.new.touch(job_path)
+  end
+
   private
 
   def unrecoverable_failure?

app/services/ci/pipeline_processing/atomic_processing_service.rb

@@ -36,6 +36,10 @@ module Ci
         update_pipeline!
         update_statuses_processed!
 
+        if Feature.enabled?(:expire_job_and_pipeline_cache_synchronously, pipeline.project, default_enabled: :yaml)
+          Ci::ExpirePipelineCacheService.new.execute(pipeline)
+        end
+
         true
       end
 

app/workers/expire_job_cache_worker.rb

@@ -15,19 +15,10 @@ class ExpireJobCacheWorker # rubocop:disable Scalability/IdempotentWorker
   idempotent!
 
   def perform(job_id)
-    job = CommitStatus.preload(:pipeline, :project).find_by_id(job_id) # rubocop: disable CodeReuse/ActiveRecord
+    job = CommitStatus.find_by_id(job_id)
     return unless job
 
-    pipeline = job.pipeline
-    project = job.project
-
-    Gitlab::EtagCaching::Store.new.touch(project_job_path(project, job))
-    ExpirePipelineCacheWorker.perform_async(pipeline.id)
-  end
-
-  private
-
-  def project_job_path(project, job)
-    Gitlab::Routing.url_helpers.project_build_path(project, job.id, format: :json)
+    job.expire_etag_cache!
+    ExpirePipelineCacheWorker.perform_async(job.pipeline_id)
   end
 end

config/feature_flags/development/expire_job_and_pipeline_cache_synchronously.yml

+---
+name: expire_job_and_pipeline_cache_synchronously
+introduced_by_url: https://gitlab.com/gitlab-org/gitlab/-/merge_requests/75611
+rollout_issue_url: https://gitlab.com/gitlab-com/gl-infra/scalability/-/issues/1380
+milestone: '14.6'
+type: development
+group: group::project management
+default_enabled: false

spec/models/ci/pipeline_spec.rb

@@ -1503,10 +1503,30 @@ RSpec.describe Ci::Pipeline, :mailer, factory_default: :keep do
     end
 
     describe 'pipeline caching' do
-      it 'performs ExpirePipelinesCacheWorker' do
-        expect(ExpirePipelineCacheWorker).to receive(:perform_async).with(pipeline.id)
+      context 'when expire_job_and_pipeline_cache_synchronously is enabled' do
+        before do
+          stub_feature_flags(expire_job_and_pipeline_cache_synchronously: true)
+        end
 
-        pipeline.cancel
+        it 'executes Ci::ExpirePipelineCacheService' do
+          expect_next_instance_of(Ci::ExpirePipelineCacheService) do |service|
+            expect(service).to receive(:execute).with(pipeline)
+          end
+
+          pipeline.cancel
+        end
+      end
+
+      context 'when expire_job_and_pipeline_cache_synchronously is disabled' do
+        before do
+          stub_feature_flags(expire_job_and_pipeline_cache_synchronously: false)
+        end
+
+        it 'performs ExpirePipelinesCacheWorker' do
+          expect(ExpirePipelineCacheWorker).to receive(:perform_async).with(pipeline.id)
+
+          pipeline.cancel
+        end
       end
     end
 

spec/models/commit_status_spec.rb

@@ -46,10 +46,28 @@ RSpec.describe CommitStatus do
   describe 'status state machine' do
     let!(:commit_status) { create(:commit_status, :running, project: project) }
 
-    it 'invalidates the cache after a transition' do
-      expect(ExpireJobCacheWorker).to receive(:perform_async).with(commit_status.id)
+    context 'when expire_job_and_pipeline_cache_synchronously is enabled' do
+      before do
+        stub_feature_flags(expire_job_and_pipeline_cache_synchronously: true)
+      end
+
+      it 'invalidates the cache after a transition' do
+        expect(commit_status).to receive(:expire_etag_cache!)
+
+        commit_status.success!
+      end
+    end
+
+    context 'when expire_job_and_pipeline_cache_synchronously is disabled' do
+      before do
+        stub_feature_flags(expire_job_and_pipeline_cache_synchronously: false)
+      end
+
+      it 'invalidates the cache after a transition' do
+        expect(ExpireJobCacheWorker).to receive(:perform_async).with(commit_status.id)
 
-      commit_status.success!
+        commit_status.success!
+      end
     end
 
     describe 'transitioning to running' do
@@ -949,4 +967,15 @@ RSpec.describe CommitStatus do
       described_class.bulk_insert_tags!(statuses, tag_list_by_build)
     end
   end
+
+  describe '#expire_etag_cache!' do
+    it 'expires the etag cache' do
+      expect_next_instance_of(Gitlab::EtagCaching::Store) do |etag_store|
+        job_path = Gitlab::Routing.url_helpers.project_build_path(project, commit_status.id, format: :json)
+        expect(etag_store).to receive(:touch).with(job_path)
+      end
+
+      commit_status.expire_etag_cache!
+    end
+  end
 end