Schedule pipeline worker only when it is uniquefix/improve-concurrent-pipeline-processing

3 files changed, 8 insertions, 17 deletions

diff --git a/app/models/commit_status.rb b/app/models/commit_status.rb
index 7b554be4f9a..4b6b996a462 100644
--- a/app/models/commit_status.rb
+++ b/app/models/commit_status.rb
@@ -90,11 +90,8 @@ class CommitStatus < ActiveRecord::Base
 
       commit_status.run_after_commit do
         pipeline.try do |pipeline|
-          if complete?
-            PipelineProcessWorker.perform_async(pipeline.id)
-          else
-            PipelineUpdateWorker.perform_async(pipeline.id)
-          end
+          worker = complete? ? PipelineProcessWorker : PipelineUpdateWorker
+          Gitlab::Worker::Unique.new(worker, pipeline.id).schedule!
         end
       end
     end
diff --git a/app/workers/pipeline_process_worker.rb b/app/workers/pipeline_process_worker.rb
index 69d2e7cf010..b89743272f7 100644
--- a/app/workers/pipeline_process_worker.rb
+++ b/app/workers/pipeline_process_worker.rb
@@ -1,13 +1,10 @@
 class PipelineProcessWorker
   include Sidekiq::Worker
-  include Gitlab::Worker::Unique
-
   sidekiq_options queue: :default
 
   def perform(pipeline_id)
-    unique_processing(pipeline_id) do
-      Ci::Pipeline.find_by(id: pipeline_id)
-        .try(:process!)
-    end
+    Gitlab::Worker::Unique.new(self.class, pipeline_id).release!
+
+    Ci::Pipeline.find_by(id: pipeline_id).try(:process!)
   end
 end
diff --git a/app/workers/pipeline_update_worker.rb b/app/workers/pipeline_update_worker.rb
index d2d22195067..e135e230cee 100644
--- a/app/workers/pipeline_update_worker.rb
+++ b/app/workers/pipeline_update_worker.rb
@@ -1,13 +1,10 @@
 class PipelineUpdateWorker
   include Sidekiq::Worker
-  include Gitlab::Worker::Unique
-
   sidekiq_options queue: :default
 
   def perform(pipeline_id)
-    unique_processing(pipeline_id) do
-      Ci::Pipeline.find_by(id: pipeline_id)
-        .try(:update_status)
-    end
+    Gitlab::Worker::Unique.new(self.class, pipeline_id).release!
+
+    Ci::Pipeline.find_by(id: pipeline_id).try(:update_status)
   end
 end