Only find projects that qualify for pull mirroring

This prevents us from needing to iterate through a bunch of mirrors that
were set up on free private projects when pull mirrors were free, and
still allows those projects to just work if they do start using a paid
plan.

ee/app/workers/update_all_mirrors_worker.rb

@@ -102,6 +102,22 @@ class UpdateAllMirrorsWorker # rubocop:disable Scalability/IdempotentWorker
 
     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 @@ class UpdateAllMirrorsWorker # rubocop:disable Scalability/IdempotentWorker
       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

ee/spec/workers/update_all_mirrors_worker_spec.rb

@@ -138,7 +138,30 @@ describe UpdateAllMirrorsWorker do
       end
     end
 
-    context 'on GitLab.com' do
+    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)
@@ -165,8 +188,35 @@ describe UpdateAllMirrorsWorker do
 
       let(:unlicensed_projects) { [unlicensed_project1, unlicensed_project2, unlicensed_project3, unlicensed_project4] }
 
-      context 'during the free period for pull mirroring' do
+      context 'after the free period for pull mirroring' do
+        before do
+          stub_feature_flags(free_period_for_pull_mirroring: false)
+        end
+
+        context 'when capacity is in excess' do
+          it 'schedules all available mirrors' do
+            schedule_mirrors!(capacity: 4)
+
+            expect_import_scheduled(licensed_project1, licensed_project2, public_project)
+            expect_import_not_scheduled(*unlicensed_projects)
+          end
+        end
+
+        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
+
+            schedule_mirrors!(capacity: 3)
+          end
+        end
+      end
+
+      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
 
@@ -174,11 +224,9 @@ describe UpdateAllMirrorsWorker do
           it "schedules all available mirrors" do
             schedule_mirrors!(capacity: 4)
 
-            aggregate_failures do
             expect_import_scheduled(licensed_project1, licensed_project2, public_project)
             expect_import_not_scheduled(*unlicensed_projects)
           end
-          end
 
           it 'requests as many batches as necessary' do
             # The first batch will only contain 3 licensed mirrors, but since we have