diff --git a/ee/app/workers/update_all_mirrors_worker.rb b/ee/app/workers/update_all_mirrors_worker.rb
index cf60b2ab822c7bfb41a12fc0713977d94514bdb2..fb94562d7e11abe08888bec6991703722d751352 100644
--- a/ee/app/workers/update_all_mirrors_worker.rb
+++ b/ee/app/workers/update_all_mirrors_worker.rb
@@ -102,6 +102,22 @@ def pull_mirrors_batch(freeze_at:, batch_size:, offset_at: nil)
 
     relation = relation.where('import_state.next_execution_timestamp > ?', offset_at) if offset_at
 
+    if check_mirror_plans_in_query?
+      root_namespaces_sql = Gitlab::ObjectHierarchy
+        .new(Namespace.where('id = projects.namespace_id'))
+        .roots
+        .select(:id)
+        .to_sql
+
+      root_namespaces_join = "INNER JOIN namespaces AS root_namespaces ON root_namespaces.id = (#{root_namespaces_sql})"
+
+      relation = relation
+        .joins(root_namespaces_join)
+        .joins('LEFT JOIN gitlab_subscriptions ON gitlab_subscriptions.namespace_id = root_namespaces.id')
+        .joins('LEFT JOIN plans ON plans.id = gitlab_subscriptions.hosted_plan_id')
+        .where(['plans.name IN (?) OR projects.visibility_level = ?', ::Plan::ALL_HOSTED_PLANS, ::Gitlab::VisibilityLevel::PUBLIC])
+    end
+
     relation
   end
   # rubocop: enable CodeReuse/ActiveRecord
@@ -113,4 +129,9 @@ def schedule_projects_in_batch(projects)
       context_proc: -> (project) { { project: project } }
     )
   end
+
+  def check_mirror_plans_in_query?
+    ::Gitlab::CurrentSettings.should_check_namespace_plan? &&
+      !::Feature.enabled?(:free_period_for_pull_mirroring, default_enabled: true)
+  end
 end
diff --git a/ee/spec/workers/update_all_mirrors_worker_spec.rb b/ee/spec/workers/update_all_mirrors_worker_spec.rb
index 9a66fb0aa0522e6c463784b2727734a430579154..900baf9d8b0077bc28ab3da339b5f7c9b230e113 100644
--- a/ee/spec/workers/update_all_mirrors_worker_spec.rb
+++ b/ee/spec/workers/update_all_mirrors_worker_spec.rb
@@ -126,7 +126,7 @@ def expect_import_not_scheduled(*projects)
       projects.each { |project| expect_import_status(project, 'none') }
     end
 
-    context 'unlicensed' do
+    context 'when the instance is unlicensed' do
       it 'does not schedule when project does not have repository mirrors available' do
         project = create(:project, :mirror)
 
@@ -138,110 +138,164 @@ def expect_import_not_scheduled(*projects)
       end
     end
 
-    context 'licensed' do
-      def scheduled_mirror(at:, licensed:)
-        namespace = create(:group, :public)
+    context 'when the instance is licensed' do
+      def scheduled_mirror(at:)
+        project = create(:project, :mirror)
+        project.import_state.update!(next_execution_timestamp: at)
+        project
+      end
+
+      before do
+        stub_feature_flags(free_period_for_pull_mirroring: false)
+      end
+
+      let_it_be(:project1) { scheduled_mirror(at: 8.weeks.ago) }
+      let_it_be(:project2) { scheduled_mirror(at: 7.weeks.ago) }
+
+      context 'when capacity is in excess' do
+        it 'schedules all available mirrors' do
+          schedule_mirrors!(capacity: 3)
+
+          expect_import_scheduled(project1, project2)
+        end
+      end
+    end
+
+    context 'when the instance checks namespace plans' do
+      def scheduled_mirror(at:, licensed:, public: false, subgroup: nil)
+        group_args = [:group, :public, subgroup && :nested].compact
+        namespace = create(*group_args)
         project = create(:project, :public, :mirror, namespace: namespace)
 
-        create(:gitlab_subscription, (licensed ? :bronze : :free), namespace: namespace)
+        create(:gitlab_subscription, (licensed ? :bronze : :free), namespace: namespace.root_ancestor)
 
         project.import_state.update!(next_execution_timestamp: at)
-        project.update!(visibility_level: Gitlab::VisibilityLevel::PRIVATE)
+        project.update!(visibility_level: Gitlab::VisibilityLevel::PRIVATE) unless public
         project
       end
 
       before do
-        stub_licensed_features(repository_mirrors: true)
         stub_application_setting(check_namespace_plan: true)
-        allow(Gitlab).to receive_messages(com?: true)
-        stub_const('License::ANY_PLAN_FEATURES', [])
       end
 
-      let!(:unlicensed_project1) { scheduled_mirror(at: 8.weeks.ago, licensed: false) }
-      let!(:unlicensed_project2) { scheduled_mirror(at: 7.weeks.ago, licensed: false) }
-      let!(:licensed_project1)   { scheduled_mirror(at: 6.weeks.ago, licensed: true) }
-      let!(:unlicensed_project3) { scheduled_mirror(at: 5.weeks.ago, licensed: false) }
-      let!(:licensed_project2)   { scheduled_mirror(at: 4.weeks.ago, licensed: true) }
-      let!(:unlicensed_project4) { scheduled_mirror(at: 3.weeks.ago, licensed: false) }
-      let!(:licensed_project3)   { scheduled_mirror(at: 1.week.ago, licensed: true) }
+      let_it_be(:unlicensed_project1) { scheduled_mirror(at: 8.weeks.ago, licensed: false) }
+      let_it_be(:unlicensed_project2) { scheduled_mirror(at: 7.weeks.ago, licensed: false) }
+      let_it_be(:licensed_project1)   { scheduled_mirror(at: 6.weeks.ago, licensed: true, subgroup: true) }
+      let_it_be(:unlicensed_project3) { scheduled_mirror(at: 5.weeks.ago, licensed: false) }
+      let_it_be(:licensed_project2)   { scheduled_mirror(at: 4.weeks.ago, licensed: true) }
+      let_it_be(:unlicensed_project4) { scheduled_mirror(at: 3.weeks.ago, licensed: false) }
+      let_it_be(:public_project)      { scheduled_mirror(at: 1.week.ago, licensed: false, public: true) }
 
       let(:unlicensed_projects) { [unlicensed_project1, unlicensed_project2, unlicensed_project3, unlicensed_project4] }
 
-      context 'when capacity is in excess' do
-        it "schedules all available mirrors" do
-          schedule_mirrors!(capacity: 4)
-
-          expect_import_scheduled(licensed_project1, licensed_project2, licensed_project3)
-          expect_import_not_scheduled(*unlicensed_projects)
+      context 'after the free period for pull mirroring' do
+        before do
+          stub_feature_flags(free_period_for_pull_mirroring: false)
         end
 
-        it 'requests as many batches as necessary' do
-          # The first batch will only contain 3 licensed mirrors, but since we have
-          # fewer than 8 mirrors in total, there's no need to request another batch
-          expect(subject).to receive(:pull_mirrors_batch).with(hash_including(batch_size: 8)).and_call_original
+        context 'when capacity is in excess' do
+          it 'schedules all available mirrors' do
+            schedule_mirrors!(capacity: 4)
 
-          schedule_mirrors!(capacity: 4)
+            expect_import_scheduled(licensed_project1, licensed_project2, public_project)
+            expect_import_not_scheduled(*unlicensed_projects)
+          end
         end
 
-        it "does not schedule a mirror of an archived project" do
-          licensed_project1.update!(archived: true)
-
-          schedule_mirrors!(capacity: 4)
+        context 'when capacity is exactly sufficient' do
+          it 'does not include unlicensed non-public projects in batches' do
+            # We expect that all three eligible projects will be
+            # included in the first batch because the query will only
+            # return eligible projects.
+            expect(subject).to receive(:pull_mirrors_batch).with(hash_including(batch_size: 6)).and_call_original.once
 
-          expect_import_scheduled(licensed_project2, licensed_project3)
-          expect_import_not_scheduled(licensed_project1)
-          expect_import_not_scheduled(*unlicensed_projects)
+            schedule_mirrors!(capacity: 3)
+          end
         end
       end
 
-      context 'when capacity is exacly sufficient' do
-        it "schedules all available mirrors" do
-          schedule_mirrors!(capacity: 3)
-
-          expect_import_scheduled(licensed_project1, licensed_project2, licensed_project3)
-          expect_import_not_scheduled(*unlicensed_projects)
+      context 'when checking licenses on each record individually' do
+        before do
+          stub_feature_flags(free_period_for_pull_mirroring: true)
+          stub_const('License::ANY_PLAN_FEATURES', [])
         end
 
-        it 'requests as many batches as necessary' do
-          # The first batch will only contain 2 licensed mirrors, so we need to request another batch
-          expect(subject).to receive(:pull_mirrors_batch).with(hash_including(batch_size: 6)).ordered.and_call_original
-          expect(subject).to receive(:pull_mirrors_batch).with(hash_including(batch_size: 2)).ordered.and_call_original
+        context 'when capacity is in excess' do
+          it "schedules all available mirrors" do
+            schedule_mirrors!(capacity: 4)
 
-          schedule_mirrors!(capacity: 3)
-        end
-      end
+            expect_import_scheduled(licensed_project1, licensed_project2, public_project)
+            expect_import_not_scheduled(*unlicensed_projects)
+          end
+
+          it 'requests as many batches as necessary' do
+            # The first batch will only contain 3 licensed mirrors, but since we have
+            # fewer than 8 mirrors in total, there's no need to request another batch
+            expect(subject).to receive(:pull_mirrors_batch).with(hash_including(batch_size: 8)).and_call_original
 
-      context 'when capacity is insufficient' do
-        it 'schedules mirrors by next_execution_timestamp' do
-          schedule_mirrors!(capacity: 2)
+            schedule_mirrors!(capacity: 4)
+          end
 
-          expect_import_scheduled(licensed_project1, licensed_project2)
-          expect_import_not_scheduled(*unlicensed_projects, licensed_project3)
+          it "does not schedule a mirror of an archived project" do
+            licensed_project1.update!(archived: true)
+
+            schedule_mirrors!(capacity: 4)
+
+            expect_import_scheduled(licensed_project2, public_project)
+            expect_import_not_scheduled(licensed_project1)
+            expect_import_not_scheduled(*unlicensed_projects)
+          end
         end
 
-        it 'requests as many batches as necessary' do
-          # The first batch will only contain 1 licensed mirror, so we need to request another batch
-          expect(subject).to receive(:pull_mirrors_batch).with(hash_including(batch_size: 4)).ordered.and_call_original
-          expect(subject).to receive(:pull_mirrors_batch).with(hash_including(batch_size: 2)).ordered.and_call_original
+        context 'when capacity is exacly sufficient' do
+          it "schedules all available mirrors" do
+            schedule_mirrors!(capacity: 3)
+
+            expect_import_scheduled(licensed_project1, licensed_project2, public_project)
+            expect_import_not_scheduled(*unlicensed_projects)
+          end
 
-          schedule_mirrors!(capacity: 2)
+          it 'requests as many batches as necessary' do
+            # The first batch will only contain 2 licensed mirrors, so we need to request another batch
+            expect(subject).to receive(:pull_mirrors_batch).with(hash_including(batch_size: 6)).ordered.and_call_original
+            expect(subject).to receive(:pull_mirrors_batch).with(hash_including(batch_size: 2)).ordered.and_call_original
+
+            schedule_mirrors!(capacity: 3)
+          end
         end
-      end
 
-      context 'when capacity is insufficient and the first batch is empty' do
-        it 'schedules mirrors by next_execution_timestamp' do
-          schedule_mirrors!(capacity: 1)
+        context 'when capacity is insufficient' do
+          it 'schedules mirrors by next_execution_timestamp' do
+            schedule_mirrors!(capacity: 2)
 
-          expect_import_scheduled(licensed_project1)
-          expect_import_not_scheduled(*unlicensed_projects, licensed_project2, licensed_project3)
+            expect_import_scheduled(licensed_project1, licensed_project2)
+            expect_import_not_scheduled(*unlicensed_projects, public_project)
+          end
+
+          it 'requests as many batches as necessary' do
+            # The first batch will only contain 1 licensed mirror, so we need to request another batch
+            expect(subject).to receive(:pull_mirrors_batch).with(hash_including(batch_size: 4)).ordered.and_call_original
+            expect(subject).to receive(:pull_mirrors_batch).with(hash_including(batch_size: 2)).ordered.and_call_original
+
+            schedule_mirrors!(capacity: 2)
+          end
         end
 
-        it 'requests as many batches as necessary' do
-          # The first batch will not contain any licensed mirrors, so we need to request another batch
-          expect(subject).to receive(:pull_mirrors_batch).with(hash_including(batch_size: 2)).ordered.and_call_original
-          expect(subject).to receive(:pull_mirrors_batch).with(hash_including(batch_size: 2)).ordered.and_call_original
+        context 'when capacity is insufficient and the first batch is empty' do
+          it 'schedules mirrors by next_execution_timestamp' do
+            schedule_mirrors!(capacity: 1)
+
+            expect_import_scheduled(licensed_project1)
+            expect_import_not_scheduled(*unlicensed_projects, licensed_project2, public_project)
+          end
+
+          it 'requests as many batches as necessary' do
+            # The first batch will not contain any licensed mirrors, so we need to request another batch
+            expect(subject).to receive(:pull_mirrors_batch).with(hash_including(batch_size: 2)).ordered.and_call_original
+            expect(subject).to receive(:pull_mirrors_batch).with(hash_including(batch_size: 2)).ordered.and_call_original
 
-          schedule_mirrors!(capacity: 1)
+            schedule_mirrors!(capacity: 1)
+          end
         end
       end
     end