diff --git a/db/post_migrate/20211023102243_schedule_delete_invalid_epic_issues.rb b/db/post_migrate/20211023102243_schedule_delete_invalid_epic_issues.rb
new file mode 100644
index 0000000000000000000000000000000000000000..cb25170f2d7c01f91435ff2956351584c77c3660
--- /dev/null
+++ b/db/post_migrate/20211023102243_schedule_delete_invalid_epic_issues.rb
@@ -0,0 +1,22 @@
+# frozen_string_literal: true
+
+class ScheduleDeleteInvalidEpicIssues < Gitlab::Database::Migration[1.0]
+  MIGRATION = 'DeleteInvalidEpicIssues'
+  DELAY_INTERVAL = 2.minutes
+  BATCH_SIZE = 10_000
+
+  disable_ddl_transaction!
+
+  def up
+    queue_background_migration_jobs_by_range_at_intervals(
+      define_batchable_model('epics'),
+      MIGRATION,
+      DELAY_INTERVAL,
+      batch_size: BATCH_SIZE,
+      track_jobs: true
+    )
+  end
+
+  def down
+  end
+end
diff --git a/db/schema_migrations/20211023102243 b/db/schema_migrations/20211023102243
new file mode 100644
index 0000000000000000000000000000000000000000..ec507da6b47d60a97cfd68885192aef7a808ba86
--- /dev/null
+++ b/db/schema_migrations/20211023102243
@@ -0,0 +1 @@
+f5039be0bd028dab4f2623fe9997a95d50bd9020ffd8b92074418024cda39b6a
\ No newline at end of file
diff --git a/ee/lib/ee/gitlab/background_migration/delete_invalid_epic_issues.rb b/ee/lib/ee/gitlab/background_migration/delete_invalid_epic_issues.rb
new file mode 100644
index 0000000000000000000000000000000000000000..792295a0cd6a4ed5a7b8123aebfa04279c42fd4f
--- /dev/null
+++ b/ee/lib/ee/gitlab/background_migration/delete_invalid_epic_issues.rb
@@ -0,0 +1,78 @@
+# frozen_string_literal: true
+
+module EE
+  module Gitlab
+    module BackgroundMigration
+      module DeleteInvalidEpicIssues
+        extend ::Gitlab::Utils::Override
+
+        class Namespace < ActiveRecord::Base
+          self.table_name = 'namespaces'
+
+          has_many :epics
+          has_many :projects
+        end
+
+        class Project < ActiveRecord::Base
+          self.table_name = 'projects'
+
+          has_many :issues
+          belongs_to :group
+        end
+
+        class Epic < ActiveRecord::Base
+          include EachBatch
+
+          self.table_name = 'epics'
+
+          has_many :epic_issues
+          belongs_to :group
+        end
+
+        class Issue < ActiveRecord::Base
+          self.table_name = 'issues'
+
+          has_many :epic_issues
+          belongs_to :project
+        end
+
+        class EpicIssue < ActiveRecord::Base
+          self.table_name = 'epic_issues'
+
+          belongs_to :epic
+          belongs_to :issue
+        end
+
+        override :perform
+        def perform(start_id, stop_id)
+          to_delete = []
+          epics = Epic.where(id: start_id..stop_id)
+            .includes(epic_issues: { issue: :project })
+            .order(:group_id)
+
+          epics.find_in_batches do |batch|
+            batch.group_by { |epic| epic.group_id }.each do |group_id, group_epics|
+              groups = group_and_hierarchy(group_id)
+
+              group_epics.each do |epic|
+                next if epic.epic_issues.empty?
+
+                epic.epic_issues.each do |epic_issue|
+                  to_delete << epic_issue.id unless groups.include?(epic_issue.issue.project.namespace_id)
+                end
+              end
+            end
+
+            EpicIssue.where(id: to_delete).delete_all if to_delete.present?
+          end
+        end
+
+        def group_and_hierarchy(id)
+          ::Gitlab::ObjectHierarchy
+            .new(Namespace.where(id: id))
+            .base_and_ancestors.pluck(:id)
+        end
+      end
+    end
+  end
+end
diff --git a/ee/spec/lib/ee/gitlab/background_migration/delete_invalid_epic_issues_spec.rb b/ee/spec/lib/ee/gitlab/background_migration/delete_invalid_epic_issues_spec.rb
new file mode 100644
index 0000000000000000000000000000000000000000..698052acb58f392c9b7b96bfdff3284810a67862
--- /dev/null
+++ b/ee/spec/lib/ee/gitlab/background_migration/delete_invalid_epic_issues_spec.rb
@@ -0,0 +1,92 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Gitlab::BackgroundMigration::DeleteInvalidEpicIssues do
+  # rubocop:disable RSpec/MultipleMemoizedHelpers
+  let!(:users) { table(:users) }
+  let!(:namespaces) { table(:namespaces) }
+  let!(:projects) { table(:projects) }
+  let!(:epics) { table(:epics) }
+  let!(:issues) { table(:issues) }
+  let!(:epic_issues) { table(:epic_issues) }
+
+  let!(:user) { users.create!(name: 'test', email: 'test@example.com', projects_limit: 5) }
+  let!(:group) { namespaces.create!(name: 'test 1', path: 'test1') }
+  let!(:sub_group) { namespaces.create!(name: 'test 2', path: 'test2', parent_id: group.id) }
+  let!(:other_group) { namespaces.create!(name: 'test 3', path: 'test3') }
+
+  let!(:project) { projects.create!(namespace_id: group.id, name: 'test 1', path: 'test1') }
+  let!(:project_sub) { projects.create!(namespace_id: sub_group.id, name: 'test 1', path: 'test1') }
+  let!(:project_other) { projects.create!(namespace_id: other_group.id, name: 'test 1', path: 'test1') }
+
+  describe '#perform' do
+    let!(:epic_before) { epics.create!(iid: 1, title: 'test 1', title_html: 'test 1', group_id: group.id, author_id: user.id) }
+    let!(:epic) { epics.create!(iid: 2, title: 'test 2', title_html: 'test 2', group_id: group.id, author_id: user.id) }
+    let!(:epic_sub) { epics.create!(iid: 3, title: 'test 3', title_html: 'test 3', group_id: sub_group.id, author_id: user.id) }
+    let!(:epic_other) { epics.create!(iid: 4, title: 'test 4', title_html: 'test 4', group_id: other_group.id, author_id: user.id) }
+    let!(:epic_last) { epics.create!(iid: 5, title: 'test 5', title_html: 'test 5', group_id: group.id, author_id: user.id) }
+
+    let!(:issue) { issues.create!(iid: 1, project_id: project.id, title: 'issue 1', title_html: 'issue 1', author_id: user.id) }
+    let!(:issue2) { issues.create!(iid: 2, project_id: project.id, title: 'issue 2', title_html: 'issue 2', author_id: user.id) }
+    let!(:issue3) { issues.create!(iid: 6, project_id: project.id, title: 'issue 3', title_html: 'issue 3', author_id: user.id) }
+    let!(:issue4) { issues.create!(iid: 7, project_id: project.id, title: 'issue 4', title_html: 'issue 4', author_id: user.id) }
+    let!(:issue5) { issues.create!(iid: 8, project_id: project.id, title: 'issue 5', title_html: 'issue 5', author_id: user.id) }
+
+    let!(:issue_sub) { issues.create!(iid: 3, project_id: project_sub.id, title: 'issue 4', title_html: 'issue 4', author_id: user.id) }
+    let!(:issue_other) { issues.create!(iid: 4, project_id: project_other.id, title: 'issue 5', title_html: 'issue 5', author_id: user.id) }
+    let!(:issue_other_2) { issues.create!(iid: 5, project_id: project_other.id, title: 'issue 6', title_html: 'issue 6', author_id: user.id) }
+    let!(:issue_other_3) { issues.create!(iid: 6, project_id: project_other.id, title: 'issue 7', title_html: 'issue 7', author_id: user.id) }
+
+    let!(:valid_and_invalid_epic_issues) do
+      invalid_epic_issues = []
+      valid_epic_issues   = []
+
+      valid_epic_issues << epic_issues.create!(issue_id: issue_other_3.id, epic_id: epic_before.id)
+      valid_epic_issues << epic_issues.create!(issue_id: issue.id, epic_id: epic_sub.id)
+      valid_epic_issues << epic_issues.create!(issue_id: issue_sub.id, epic_id: epic_sub.id)
+      invalid_epic_issues << epic_issues.create!(issue_id: issue_other.id, epic_id: epic_sub.id)
+      valid_epic_issues << epic_issues.create!(issue_id: issue_other_2.id, epic_id: epic_other.id)
+      invalid_epic_issues << epic_issues.create!(issue_id: issue2.id, epic_id: epic_other.id)
+      valid_epic_issues << epic_issues.create!(issue_id: issue3.id, epic_id: epic.id)
+      valid_epic_issues << epic_issues.create!(issue_id: issue5.id, epic_id: epic_last.id)
+
+      { valid: valid_epic_issues, invalid: invalid_epic_issues }
+    end
+
+    let(:valid_epic_issues) { valid_and_invalid_epic_issues[:valid] }
+    let(:invalid_epic_issues) { valid_and_invalid_epic_issues[:invalid] }
+
+    it 'removes invalid epic issues' do
+      expect { described_class.new.perform(epic.id, epic_last.id) }.to change { epic_issues.count }.from(8).to(6)
+
+      expect(epic_issues.all).to match_array(valid_epic_issues)
+    end
+
+    it 'searches the group hierarchy only once per epics in the same group' do
+      service = described_class.new
+
+      expect(service).to receive(:group_and_hierarchy).exactly(3).times.and_call_original
+
+      service.perform(epic.id, epic_last.id)
+    end
+
+    it 'prevents N+1 queries' do
+      control = ActiveRecord::QueryRecorder.new(skip_cached: false) do
+        described_class.new.perform(epic.id, epic_other.id)
+      end
+
+      # recreate deleted records
+      epic_issues.create!(issue_id: issue_other.id, epic_id: epic_sub.id)
+      epic_issues.create!(issue_id: issue2.id, epic_id: epic_other.id)
+
+      # create new records to delete
+      issue_9 = issues.create!(iid: 9, project_id: project.id, title: 'issue 7', title_html: 'issue 7', author_id: user.id)
+      issue_10 = issues.create!(iid: 10, project_id: project.id, title: 'issue 8', title_html: 'issue 8', author_id: user.id)
+      epic_issues.create!(issue_id: issue_9.id, epic_id: epic_other.id)
+      epic_issues.create!(issue_id: issue_10.id, epic_id: epic_other.id)
+
+      expect { described_class.new.perform(epic.id, epic_other.id) }.not_to exceed_all_query_limit(control)
+    end
+  end
+end
diff --git a/ee/spec/migrations/schedule_delete_invalid_epic_issues_spec.rb b/ee/spec/migrations/schedule_delete_invalid_epic_issues_spec.rb
new file mode 100644
index 0000000000000000000000000000000000000000..67184f52071c5309438c900d972f87a2ebb8f217
--- /dev/null
+++ b/ee/spec/migrations/schedule_delete_invalid_epic_issues_spec.rb
@@ -0,0 +1,32 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+require_migration!
+
+RSpec.describe ScheduleDeleteInvalidEpicIssues do
+  let(:migration) { described_class::MIGRATION }
+  let(:users) { table(:users) }
+  let(:namespaces) { table(:namespaces) }
+  let(:epics) { table(:epics) }
+
+  let(:user) { users.create!(name: 'test', email: 'test@example.com', projects_limit: 5) }
+  let!(:group) { namespaces.create!(name: 'test 1', path: 'test1') }
+
+  let!(:epic1) { epics.create!(iid: 1, title: 'test 1', title_html: 'test 1', group_id: group.id, author_id: user.id) }
+  let!(:epic2) { epics.create!(iid: 2, title: 'test 2', title_html: 'test 2', group_id: group.id, author_id: user.id) }
+  let!(:epic3) { epics.create!(iid: 3, title: 'test 3', title_html: 'test 3', group_id: group.id, author_id: user.id) }
+
+  it 'correctly schedules background migrations' do
+    stub_const("#{described_class.name}::BATCH_SIZE", 2)
+
+    Sidekiq::Testing.fake! do
+      freeze_time do
+        migrate!
+
+        expect(BackgroundMigrationWorker.jobs.size).to eq(2)
+        expect(migration).to be_scheduled_delayed_migration(2.minutes, epic1.id, epic2.id)
+        expect(migration).to be_scheduled_delayed_migration(4.minutes, epic3.id, epic3.id)
+      end
+    end
+  end
+end
diff --git a/lib/gitlab/background_migration/delete_invalid_epic_issues.rb b/lib/gitlab/background_migration/delete_invalid_epic_issues.rb
new file mode 100644
index 0000000000000000000000000000000000000000..549c3b4fc8cad8a8254ede1790fa018cd6c21d16
--- /dev/null
+++ b/lib/gitlab/background_migration/delete_invalid_epic_issues.rb
@@ -0,0 +1,13 @@
+# frozen_string_literal: true
+
+module Gitlab
+  module BackgroundMigration
+    # rubocop: disable Style/Documentation
+    class DeleteInvalidEpicIssues
+      def perform(start_id, stop_id)
+      end
+    end
+  end
+end
+
+Gitlab::BackgroundMigration::DeleteInvalidEpicIssues.prepend_mod_with('Gitlab::BackgroundMigration::DeleteInvalidEpicIssues')