diff --git a/app/services/ci/ensure_stage_service.rb b/app/services/ci/ensure_stage_service.rb
index dc2f49e8db1..87f19b333de 100644
--- a/app/services/ci/ensure_stage_service.rb
+++ b/app/services/ci/ensure_stage_service.rb
@@ -7,6 +7,8 @@ module Ci
   # stage.
   #
   class EnsureStageService < BaseService
+    EnsureStageError = Class.new(StandardError)
+
     def execute(build)
       @build = build
 
@@ -22,8 +24,16 @@ module Ci
 
     private
 
-    def ensure_stage
+    def ensure_stage(attempts: 2)
       find_stage || create_stage
+    rescue ActiveRecord::RecordNotUnique
+      retry if (attempts -= 1) > 0
+
+      raise EnsureStageError, <<~EOS
+        We failed to find or create a unique pipeline stage after 2 retries.
+        This should never happen and is most likely the result of a bug in
+        the database load balancing code.
+      EOS
     end
 
     def find_stage
diff --git a/app/services/ci/retry_build_service.rb b/app/services/ci/retry_build_service.rb
index c552193e66b..6128b2a8fbb 100644
--- a/app/services/ci/retry_build_service.rb
+++ b/app/services/ci/retry_build_service.rb
@@ -1,7 +1,7 @@
 module Ci
   class RetryBuildService < ::BaseService
     CLONE_ACCESSORS = %i[pipeline project ref tag options commands name
-                         allow_failure stage_id stage stage_idx trigger_request
+                         allow_failure stage stage_id stage_idx trigger_request
                          yaml_variables when environment coverage_regex
                          description tag_list protected].freeze
 
diff --git a/db/post_migrate/20180119121225_remove_redundant_pipeline_stages.rb b/db/post_migrate/20180119121225_remove_redundant_pipeline_stages.rb
new file mode 100644
index 00000000000..61ea85eb2a7
--- /dev/null
+++ b/db/post_migrate/20180119121225_remove_redundant_pipeline_stages.rb
@@ -0,0 +1,66 @@
+class RemoveRedundantPipelineStages < ActiveRecord::Migration
+  include Gitlab::Database::MigrationHelpers
+
+  DOWNTIME = false
+
+  disable_ddl_transaction!
+
+  def up(attempts: 100)
+    remove_redundant_pipeline_stages!
+    remove_outdated_index!
+    add_unique_index!
+  rescue ActiveRecord::RecordNotUnique
+    retry if (attempts -= 1) > 0
+
+    raise StandardError, <<~EOS
+      Failed to add an unique index to ci_stages, despite retrying the
+      migration 100 times.
+
+      See https://gitlab.com/gitlab-org/gitlab-ce/merge_requests/16580.
+    EOS
+  end
+
+  def down
+    remove_concurrent_index :ci_stages, [:pipeline_id, :name], unique: true
+    add_concurrent_index :ci_stages, [:pipeline_id, :name]
+  end
+
+  private
+
+  def remove_outdated_index!
+    return unless index_exists?(:ci_stages, [:pipeline_id, :name])
+
+    remove_concurrent_index :ci_stages, [:pipeline_id, :name]
+  end
+
+  def add_unique_index!
+    add_concurrent_index :ci_stages, [:pipeline_id, :name], unique: true
+  end
+
+  def remove_redundant_pipeline_stages!
+    disable_statement_timeout
+
+    redundant_stages_ids = <<~SQL
+      SELECT id FROM ci_stages WHERE (pipeline_id, name) IN (
+        SELECT pipeline_id, name FROM ci_stages
+          GROUP BY pipeline_id, name HAVING COUNT(*) > 1
+      )
+    SQL
+
+    execute <<~SQL
+      UPDATE ci_builds SET stage_id = NULL WHERE stage_id IN (#{redundant_stages_ids})
+    SQL
+
+    if Gitlab::Database.postgresql?
+      execute <<~SQL
+        DELETE FROM ci_stages WHERE id IN (#{redundant_stages_ids})
+      SQL
+    else # We can't modify a table we are selecting from on MySQL
+      execute <<~SQL
+        DELETE a FROM ci_stages AS a, ci_stages AS b
+          WHERE a.pipeline_id = b.pipeline_id AND a.name = b.name
+            AND a.id <> b.id
+      SQL
+    end
+  end
+end
diff --git a/db/schema.rb b/db/schema.rb
index 7282ecfc3e7..9ab4dabfb7d 100644
--- a/db/schema.rb
+++ b/db/schema.rb
@@ -451,7 +451,7 @@ ActiveRecord::Schema.define(version: 20180206200543) do
     t.integer "lock_version"
   end
 
-  add_index "ci_stages", ["pipeline_id", "name"], name: "index_ci_stages_on_pipeline_id_and_name", using: :btree
+  add_index "ci_stages", ["pipeline_id", "name"], name: "index_ci_stages_on_pipeline_id_and_name", unique: true, using: :btree
   add_index "ci_stages", ["pipeline_id"], name: "index_ci_stages_on_pipeline_id", using: :btree
   add_index "ci_stages", ["project_id"], name: "index_ci_stages_on_project_id", using: :btree
 
diff --git a/spec/migrations/remove_redundant_pipeline_stages_spec.rb b/spec/migrations/remove_redundant_pipeline_stages_spec.rb
new file mode 100644
index 00000000000..8325f986594
--- /dev/null
+++ b/spec/migrations/remove_redundant_pipeline_stages_spec.rb
@@ -0,0 +1,59 @@
+require 'spec_helper'
+require Rails.root.join('db', 'post_migrate', '20180119121225_remove_redundant_pipeline_stages.rb')
+
+describe RemoveRedundantPipelineStages, :migration do
+  let(:projects) { table(:projects) }
+  let(:pipelines) { table(:ci_pipelines) }
+  let(:stages) { table(:ci_stages) }
+  let(:builds) { table(:ci_builds) }
+
+  before do
+    projects.create!(id: 123, name: 'gitlab', path: 'gitlab-ce')
+    pipelines.create!(id: 234, project_id: 123, ref: 'master', sha: 'adf43c3a')
+
+    stages.create!(id: 6, project_id: 123, pipeline_id: 234, name: 'build')
+    stages.create!(id: 10, project_id: 123, pipeline_id: 234, name: 'build')
+    stages.create!(id: 21, project_id: 123, pipeline_id: 234, name: 'build')
+    stages.create!(id: 41, project_id: 123, pipeline_id: 234, name: 'test')
+    stages.create!(id: 62, project_id: 123, pipeline_id: 234, name: 'test')
+    stages.create!(id: 102, project_id: 123, pipeline_id: 234, name: 'deploy')
+
+    builds.create!(id: 1, commit_id: 234, project_id: 123, stage_id: 10)
+    builds.create!(id: 2, commit_id: 234, project_id: 123, stage_id: 21)
+    builds.create!(id: 3, commit_id: 234, project_id: 123, stage_id: 21)
+    builds.create!(id: 4, commit_id: 234, project_id: 123, stage_id: 41)
+    builds.create!(id: 5, commit_id: 234, project_id: 123, stage_id: 62)
+    builds.create!(id: 6, commit_id: 234, project_id: 123, stage_id: 102)
+  end
+
+  it 'removes ambiguous stages and preserves builds' do
+    expect(stages.all.count).to eq 6
+    expect(builds.all.count).to eq 6
+
+    migrate!
+
+    expect(stages.all.count).to eq 1
+    expect(builds.all.count).to eq 6
+    expect(builds.all.pluck(:stage_id).compact).to eq [102]
+  end
+
+  it 'retries when incorrectly added index exception is caught' do
+    allow_any_instance_of(described_class)
+      .to receive(:remove_redundant_pipeline_stages!)
+
+    expect_any_instance_of(described_class)
+      .to receive(:remove_outdated_index!)
+      .exactly(100).times.and_call_original
+
+    expect { migrate! }
+      .to raise_error StandardError, /Failed to add an unique index/
+  end
+
+  it 'does not retry when unknown exception is being raised' do
+    allow(subject).to receive(:remove_outdated_index!)
+    expect(subject).to receive(:remove_redundant_pipeline_stages!).once
+    allow(subject).to receive(:add_unique_index!).and_raise(StandardError)
+
+    expect { subject.up(attempts: 3) }.to raise_error StandardError
+  end
+end
diff --git a/spec/services/ci/ensure_stage_service_spec.rb b/spec/services/ci/ensure_stage_service_spec.rb
new file mode 100644
index 00000000000..d17e30763d7
--- /dev/null
+++ b/spec/services/ci/ensure_stage_service_spec.rb
@@ -0,0 +1,51 @@
+require 'spec_helper'
+
+describe Ci::EnsureStageService, '#execute' do
+  set(:project) { create(:project) }
+  set(:user) { create(:user) }
+
+  let(:stage) { create(:ci_stage_entity) }
+  let(:job) { build(:ci_build) }
+
+  let(:service) { described_class.new(project, user) }
+
+  context 'when build has a stage assigned' do
+    it 'does not create a new stage' do
+      job.assign_attributes(stage_id: stage.id)
+
+      expect { service.execute(job) }.not_to change { Ci::Stage.count }
+    end
+  end
+
+  context 'when build does not have a stage assigned' do
+    it 'creates a new stage' do
+      job.assign_attributes(stage_id: nil, stage: 'test')
+
+      expect { service.execute(job) }.to change { Ci::Stage.count }.by(1)
+    end
+  end
+
+  context 'when build is invalid' do
+    it 'does not create a new stage' do
+      job.assign_attributes(stage_id: nil, ref: nil)
+
+      expect { service.execute(job) }.not_to change { Ci::Stage.count }
+    end
+  end
+
+  context 'when new stage can not be created because of an exception' do
+    before do
+      allow(Ci::Stage).to receive(:create!)
+        .and_raise(ActiveRecord::RecordNotUnique.new('Duplicates!'))
+    end
+
+    it 'retries up to two times' do
+      job.assign_attributes(stage_id: nil)
+
+      expect(service).to receive(:find_stage).exactly(2).times
+
+      expect { service.execute(job) }
+        .to raise_error(Ci::EnsureStageService::EnsureStageError)
+    end
+  end
+end
diff --git a/spec/services/ci/retry_build_service_spec.rb b/spec/services/ci/retry_build_service_spec.rb
index 2c2f48e323d..db9c216d3f4 100644
--- a/spec/services/ci/retry_build_service_spec.rb
+++ b/spec/services/ci/retry_build_service_spec.rb
@@ -5,7 +5,11 @@ describe Ci::RetryBuildService do
   set(:project) { create(:project) }
   set(:pipeline) { create(:ci_pipeline, project: project) }
 
-  let(:build) { create(:ci_build, pipeline: pipeline) }
+  let(:stage) do
+    Ci::Stage.create!(project: project, pipeline: pipeline, name: 'test')
+  end
+
+  let(:build) { create(:ci_build, pipeline: pipeline, stage_id: stage.id) }
 
   let(:service) do
     described_class.new(project, user)
@@ -27,29 +31,27 @@ describe Ci::RetryBuildService do
        user_id auto_canceled_by_id retried failure_reason].freeze
 
   shared_examples 'build duplication' do
-    let(:stage) do
-      # TODO, we still do not have factory for new stages, we will need to
-      # switch existing factory to persist stages, instead of using LegacyStage
-      #
-      Ci::Stage.create!(project: project, pipeline: pipeline, name: 'test')
-    end
+    let(:another_pipeline) { create(:ci_empty_pipeline, project: project) }
 
     let(:build) do
       create(:ci_build, :failed, :artifacts, :expired, :erased,
              :queued, :coverage, :tags, :allowed_to_fail, :on_tag,
              :triggered, :trace_artifact, :teardown_environment,
-             description: 'my-job', stage: 'test',  pipeline: pipeline,
-             auto_canceled_by: create(:ci_empty_pipeline, project: project)) do |build|
-               ##
-               # TODO, workaround for FactoryBot limitation when having both
-               # stage (text) and stage_id (integer) columns in the table.
-               build.stage_id = stage.id
-             end
+             description: 'my-job', stage: 'test', stage_id: stage.id,
+             pipeline: pipeline, auto_canceled_by: another_pipeline)
+    end
+
+    before do
+      # Make sure that build has both `stage_id` and `stage` because FactoryBot
+      # can reset one of the fields when assigning another. We plan to deprecate
+      # and remove legacy `stage` column in the future.
+      build.update_attributes(stage: 'test', stage_id: stage.id)
     end
 
     describe 'clone accessors' do
       CLONE_ACCESSORS.each do |attribute|
         it "clones #{attribute} build attribute" do
+          expect(build.send(attribute)).not_to be_nil
           expect(new_build.send(attribute)).not_to be_nil
           expect(new_build.send(attribute)).to eq build.send(attribute)
         end
@@ -122,10 +124,12 @@ describe Ci::RetryBuildService do
 
       context 'when there are subsequent builds that are skipped' do
         let!(:subsequent_build) do
-          create(:ci_build, :skipped, stage_idx: 1, pipeline: pipeline)
+          create(:ci_build, :skipped, stage_idx: 2,
+                                      pipeline: pipeline,
+                                      stage: 'deploy')
         end
 
-        it 'resumes pipeline processing in subsequent stages' do
+        it 'resumes pipeline processing in a subsequent stage' do
           service.execute(build)
 
           expect(subsequent_build.reload).to be_created