Add options hash param to CancelRedundantPipelinesWorker

app/workers/ci/cancel_redundant_pipelines_worker.rb

@@ -10,11 +10,12 @@ class CancelRedundantPipelinesWorker
     deduplicate :until_executed
     urgency :high
-    def perform(pipeline_id)
-      Ci::Pipeline.find_by_id(pipeline_id).try do |pipeline|
-        Ci::PipelineCreation::CancelRedundantPipelinesService
-          .new(pipeline)
-          .execute
+    def perform(pipeline_id, options = {})
+      relation = Ci::Pipeline.all
+      relation = relation.in_partition(options['partition_id']) if options['partition_id'].present?
+
+      relation.find_by_id(pipeline_id).try do |pipeline|
+        Ci::PipelineCreation::CancelRedundantPipelinesService.new(pipeline).execute
       end
     end
   end

spec/workers/ci/cancel_redundant_pipelines_worker_spec.rb

@@ -24,6 +24,24 @@
       perform
     end
+    context 'with a passed partition_id' do
+      subject(:perform) { described_class.new.perform(pipeline.id, { 'partition_id' => pipeline.partition_id }) }
+
+      it 'finds the passed pipeline in the correct partition and passes it to the service' do
+        expect(Ci::PipelineCreation::CancelRedundantPipelinesService)
+          .to receive(:new)
+          .with(pipeline)
+          .and_return(service)
+
+        expect(service).to receive(:execute)
+
+        recorder = ActiveRecord::QueryRecorder.new { perform }
+
+        expect(recorder.count).to eq(1)
+        expect(recorder.log.first).to match(/^SELECT "ci_pipelines".*"partition_id" = #{pipeline.partition_id}/)
+      end
+    end
+
     context 'if pipeline is deleted' do
       subject(:perform) { described_class.new.perform(non_existing_record_id) }