diff --git a/db/docs/batched_background_migrations/delete_orphaned_stage_records.yml b/db/docs/batched_background_migrations/delete_orphaned_stage_records.yml
new file mode 100644
index 0000000000000000000000000000000000000000..9b9b1de53e6cc4e6ccdcd193d2e681ed7d0d7d96
--- /dev/null
+++ b/db/docs/batched_background_migrations/delete_orphaned_stage_records.yml
@@ -0,0 +1,8 @@
+---
+migration_job_name: DeleteOrphanedStageRecords
+description: Delete corrupted rows from p_ci_stages
+feature_category: continuous_integration
+introduced_by_url: https://gitlab.com/gitlab-org/gitlab/-/merge_requests/168750
+milestone: '17.6'
+queued_migration_version: 20241009135743
+finalized_by: # version of the migration that finalized this BBM
diff --git a/db/post_migrate/20241009135743_queue_delete_orphaned_stage_records.rb b/db/post_migrate/20241009135743_queue_delete_orphaned_stage_records.rb
new file mode 100644
index 0000000000000000000000000000000000000000..42b0e5b9fcb69a4fdce2a8b0f03125915b192baa
--- /dev/null
+++ b/db/post_migrate/20241009135743_queue_delete_orphaned_stage_records.rb
@@ -0,0 +1,28 @@
+# frozen_string_literal: true
+
+class QueueDeleteOrphanedStageRecords < Gitlab::Database::Migration[2.2]
+ milestone '17.6'
+
+ restrict_gitlab_migration gitlab_schema: :gitlab_ci
+
+ MIGRATION = "DeleteOrphanedStageRecords"
+ DELAY_INTERVAL = 2.minutes
+ BATCH_SIZE = 1000
+ SUB_BATCH_SIZE = 100
+
+ def up
+ queue_batched_background_migration(
+ MIGRATION,
+ :p_ci_stages,
+ :pipeline_id,
+ job_interval: DELAY_INTERVAL,
+ batch_size: BATCH_SIZE,
+ batch_class_name: 'LooseIndexScanBatchingStrategy',
+ sub_batch_size: SUB_BATCH_SIZE
+ )
+ end
+
+ def down
+ delete_batched_background_migration(MIGRATION, :p_ci_stages, :pipeline_id, [])
+ end
+end
diff --git a/db/schema_migrations/20241009135743 b/db/schema_migrations/20241009135743
new file mode 100644
index 0000000000000000000000000000000000000000..462f673c9ecfa56b6b886fa795bc04e9c69c0c2a
--- /dev/null
+++ b/db/schema_migrations/20241009135743
@@ -0,0 +1 @@
+84bd92cf4afd4e72c708ac6b81978928ae560e6eb4a36d96678d0eaf9e51bce4
\ No newline at end of file
diff --git a/lib/gitlab/background_migration/delete_orphaned_stage_records.rb b/lib/gitlab/background_migration/delete_orphaned_stage_records.rb
new file mode 100644
index 0000000000000000000000000000000000000000..ea92c640ef8101aa49e9c519727529803e3ba31b
--- /dev/null
+++ b/lib/gitlab/background_migration/delete_orphaned_stage_records.rb
@@ -0,0 +1,37 @@
+# frozen_string_literal: true
+
+module Gitlab
+ module BackgroundMigration
+ class DeleteOrphanedStageRecords < BatchedMigrationJob
+ operation_name :delete_orphaned_stage_records
+ feature_category :continuous_integration
+
+ class CiPipeline < ::Ci::ApplicationRecord
+ self.table_name = :p_ci_pipelines
+ self.primary_key = :id
+ end
+
+ def perform
+ distinct_each_batch do |batch|
+ pipeline_ids = batch.pluck(batch_column)
+ pipelines_query = CiPipeline
+ .where('p_ci_stages.pipeline_id = p_ci_pipelines.id')
+ .where('p_ci_stages.partition_id = p_ci_pipelines.partition_id')
+ .select(1)
+
+ base_relation
+ .where(batch_column => pipeline_ids)
+ .where('NOT EXISTS (?)', pipelines_query)
+ .delete_all
+ end
+ end
+
+ private
+
+ def base_relation
+ define_batchable_model(batch_table, connection: connection, primary_key: :id)
+ .where(batch_column => start_id..end_id)
+ end
+ end
+ end
+end
diff --git a/spec/lib/gitlab/background_migration/delete_orphaned_stage_records_spec.rb b/spec/lib/gitlab/background_migration/delete_orphaned_stage_records_spec.rb
new file mode 100644
index 0000000000000000000000000000000000000000..fc6accee021073fde59aa5cc8b9e4f25f6c18058
--- /dev/null
+++ b/spec/lib/gitlab/background_migration/delete_orphaned_stage_records_spec.rb
@@ -0,0 +1,61 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Gitlab::BackgroundMigration::DeleteOrphanedStageRecords,
+ feature_category: :continuous_integration, migration: :gitlab_ci do
+ let(:pipelines_table) { table(:p_ci_pipelines, database: :ci, primary_key: :id) }
+ let(:stages_table) { table(:p_ci_stages, database: :ci, primary_key: :id) }
+
+ let(:default_attributes) { { project_id: 600, partition_id: 100 } }
+ let!(:regular_pipeline) { pipelines_table.create!(default_attributes) }
+ let!(:deleted_pipeline) { pipelines_table.create!(default_attributes) }
+ let!(:other_pipeline) { pipelines_table.create!(default_attributes) }
+
+ let!(:regular_build) do
+ stages_table.create!(pipeline_id: regular_pipeline.id, **default_attributes)
+ end
+
+ let!(:orphaned_build) do
+ stages_table.create!(pipeline_id: deleted_pipeline.id, **default_attributes)
+ end
+
+ let(:connection) { Ci::ApplicationRecord.connection }
+
+ around do |example|
+ connection.transaction do
+ connection.execute(<<~SQL)
+ ALTER TABLE ci_pipelines DISABLE TRIGGER ALL;
+ SQL
+
+ example.run
+
+ connection.execute(<<~SQL)
+ ALTER TABLE ci_pipelines ENABLE TRIGGER ALL;
+ SQL
+ end
+ end
+
+ describe '#perform' do
+ subject(:migration) do
+ described_class.new(
+ start_id: stages_table.minimum(:pipeline_id),
+ end_id: stages_table.maximum(:pipeline_id),
+ batch_table: :p_ci_stages,
+ batch_column: :pipeline_id,
+ sub_batch_size: 100,
+ pause_ms: 0,
+ connection: connection
+ )
+ end
+
+ it 'deletes from p_ci_stages where pipeline_id has no related record at p_ci_pipelines.id', :aggregate_failures do
+ expect { deleted_pipeline.delete }.to not_change { stages_table.count }
+
+ expect { migration.perform }.to change { stages_table.count }.from(2).to(1)
+
+ expect(regular_build.reload).to be_persisted
+ expect { orphaned_build.reload }.to raise_error(ActiveRecord::RecordNotFound)
+ end
+ end
+end
diff --git a/spec/migrations/20241009135743_queue_delete_orphaned_stage_records_spec.rb b/spec/migrations/20241009135743_queue_delete_orphaned_stage_records_spec.rb
new file mode 100644
index 0000000000000000000000000000000000000000..176d97c55180a97a970b39b30436c36b34fa8bea
--- /dev/null
+++ b/spec/migrations/20241009135743_queue_delete_orphaned_stage_records_spec.rb
@@ -0,0 +1,27 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+require_migration!
+
+RSpec.describe QueueDeleteOrphanedStageRecords, migration: :gitlab_ci, feature_category: :continuous_integration do
+ let!(:batched_migration) { described_class::MIGRATION }
+
+ it 'schedules a new batched migration' do
+ reversible_migration do |migration|
+ migration.before -> {
+ expect(batched_migration).not_to have_scheduled_batched_migration
+ }
+
+ migration.after -> {
+ expect(batched_migration).to have_scheduled_batched_migration(
+ table_name: :p_ci_stages,
+ column_name: :pipeline_id,
+ interval: described_class::DELAY_INTERVAL,
+ batch_size: described_class::BATCH_SIZE,
+ sub_batch_size: described_class::SUB_BATCH_SIZE,
+ gitlab_schema: :gitlab_ci
+ )
+ }
+ end
+ end
+end