1 files changed, 29 insertions, 3 deletions
diff --git a/app/services/ci/pipeline_artifacts/destroy_all_expired_service.rb b/app/services/ci/pipeline_artifacts/destroy_all_expired_service.rb
index 17c039885e5..8dddf3c3f6c 100644
--- a/app/services/ci/pipeline_artifacts/destroy_all_expired_service.rb
+++ b/app/services/ci/pipeline_artifacts/destroy_all_expired_service.rb
@@ -3,20 +3,26 @@
 module Ci
   module PipelineArtifacts
     class DestroyAllExpiredService
+      include ::Gitlab::ExclusiveLeaseHelpers
       include ::Gitlab::LoopHelpers
       include ::Gitlab::Utils::StrongMemoize
 
       BATCH_SIZE = 100
-      LOOP_TIMEOUT = 5.minutes
       LOOP_LIMIT = 1000
+      LOOP_TIMEOUT = 5.minutes
+      LOCK_TIMEOUT = 10.minutes
+      EXCLUSIVE_LOCK_KEY = 'expired_pipeline_artifacts:destroy:lock'
 
       def initialize
         @removed_artifacts_count = 0
+        @start_at = Time.current
       end
 
       def execute
-        loop_until(timeout: LOOP_TIMEOUT, limit: LOOP_LIMIT) do
-          destroy_artifacts_batch
+        in_lock(EXCLUSIVE_LOCK_KEY, ttl: LOCK_TIMEOUT, retries: 1) do
+          destroy_unlocked_pipeline_artifacts
+
+          legacy_destroy_pipeline_artifacts
         end
 
         @removed_artifacts_count
@@ -24,10 +30,30 @@ module Ci
 
       private
 
+      def destroy_unlocked_pipeline_artifacts
+        loop_until(timeout: LOOP_TIMEOUT, limit: LOOP_LIMIT) do
+          artifacts = Ci::PipelineArtifact.expired_before(@start_at).artifact_unlocked.limit(BATCH_SIZE)
+
+          break if artifacts.empty?
+
+          destroy_batch(artifacts)
+        end
+      end
+
+      def legacy_destroy_pipeline_artifacts
+        loop_until(timeout: LOOP_TIMEOUT, limit: LOOP_LIMIT) do
+          destroy_artifacts_batch
+        end
+      end
+
       def destroy_artifacts_batch
         artifacts = ::Ci::PipelineArtifact.unlocked.expired.limit(BATCH_SIZE).to_a
         return false if artifacts.empty?
 
+        destroy_batch(artifacts)
+      end
+
+      def destroy_batch(artifacts)
         artifacts.each(&:destroy!)
         increment_stats(artifacts.size)