Remove legacy query to find job artifacts that have been synced

This commit removes the legacy queries to find job artifacts
that have been synced since we made Foreign Data Wrapper (FDW)
a hard requirement for Geo on GitLab 12.0.

ee/app/finders/geo/job_artifact_registry_finder.rb

@@ -7,7 +7,7 @@ module Geo
     end
 
     def count_synced
-      job_artifacts_synced.count
+      registries_for_job_artifacts.merge(Geo::JobArtifactRegistry.synced).count
     end
 
     def count_failed
@@ -112,14 +112,6 @@ module Geo
       end
     end
 
-    def job_artifacts_synced
-      if use_legacy_queries_for_selective_sync?
-        legacy_finder.job_artifacts_synced
-      else
-        registries_for_job_artifacts.merge(Geo::JobArtifactRegistry.synced)
-      end
-    end
-
     def job_artifacts_failed
       if use_legacy_queries_for_selective_sync?
         legacy_finder.job_artifacts_failed

ee/app/finders/geo/legacy_job_artifact_registry_finder.rb

@@ -6,14 +6,6 @@ module Geo
       current_node.job_artifacts.syncable
     end
 
-    def job_artifacts_synced
-      legacy_inner_join_registry_ids(
-        syncable,
-        Geo::JobArtifactRegistry.synced.pluck_artifact_key,
-        Ci::JobArtifact
-      )
-    end
-
     def job_artifacts_failed
       legacy_inner_join_registry_ids(
         syncable,

ee/spec/finders/geo/job_artifact_registry_finder_spec.rb

@@ -563,5 +563,342 @@ describe Geo::JobArtifactRegistryFinder, :geo do
     end
   end
 
-  it_behaves_like 'a file registry finder'
+  it 'responds to file registry finder methods' do
+    file_registry_finder_methods = %i{
+      syncable
+      count_syncable
+      count_synced
+      count_failed
+      count_synced_missing_on_primary
+      count_registry
+      find_unsynced
+      find_migrated_local
+      find_retryable_failed_registries
+      find_retryable_synced_missing_on_primary_registries
+    }
+
+    file_registry_finder_methods.each do |method|
+      expect(subject).to respond_to(method)
+    end
+  end
+
+  context 'FDW', :geo_fdw do
+    context 'with use_fdw_queries_for_selective_sync disabled' do
+      before do
+        stub_feature_flags(use_fdw_queries_for_selective_sync: false)
+      end
+
+      include_examples 'counts all the things'
+      include_examples 'finds all the things'
+    end
+
+    context 'with use_fdw_queries_for_selective_sync enabled' do
+      before do
+        stub_feature_flags(use_fdw_queries_for_selective_sync: true)
+      end
+
+      include_examples 'counts all the things'
+      include_examples 'finds all the things'
+    end
+  end
+
+  context 'Legacy' do
+    before do
+      stub_fdw_disabled
+    end
+
+    describe '#count_syncable' do
+      let!(:job_artifact_1) { create(:ci_job_artifact, project: synced_project) }
+      let!(:job_artifact_2) { create(:ci_job_artifact, project: unsynced_project) }
+      let!(:job_artifact_3) { create(:ci_job_artifact, project: synced_project) }
+      let!(:job_artifact_4) { create(:ci_job_artifact, project: unsynced_project) }
+      let!(:job_artifact_5) { create(:ci_job_artifact, project: project_broken_storage) }
+      let!(:job_artifact_6) { create(:ci_job_artifact, project: project_broken_storage) }
+
+      it 'counts job artifacts' do
+        expect(subject.count_syncable).to eq 6
+      end
+
+      it 'ignores remote job artifacts' do
+        job_artifact_1.update_column(:file_store, ObjectStorage::Store::REMOTE)
+
+        expect(subject.count_syncable).to eq 5
+      end
+
+      it 'ignores expired job artifacts' do
+        job_artifact_1.update_column(:expire_at, Date.yesterday)
+
+        expect(subject.count_syncable).to eq 5
+      end
+
+      context 'with selective sync by namespace' do
+        before do
+          secondary.update!(selective_sync_type: 'namespaces', namespaces: [synced_group])
+        end
+
+        it 'counts job artifacts' do
+          expect(subject.count_syncable).to eq 2
+        end
+
+        it 'ignores remote job artifacts' do
+          job_artifact_1.update_column(:file_store, ObjectStorage::Store::REMOTE)
+
+          expect(subject.count_syncable).to eq 1
+        end
+
+        it 'ignores expired job artifacts' do
+          job_artifact_1.update_column(:expire_at, Date.yesterday)
+
+          expect(subject.count_syncable).to eq 1
+        end
+      end
+
+      context 'with selective sync by shard' do
+        before do
+          secondary.update!(selective_sync_type: 'shards', selective_sync_shards: ['broken'])
+        end
+
+        it 'counts job artifacts' do
+          expect(subject.count_syncable).to eq 2
+        end
+
+        it 'ignores remote job artifacts' do
+          job_artifact_5.update_column(:file_store, ObjectStorage::Store::REMOTE)
+
+          expect(subject.count_syncable).to eq 1
+        end
+
+        it 'ignores expired job artifacts' do
+          job_artifact_5.update_column(:expire_at, Date.yesterday)
+
+          expect(subject.count_syncable).to eq 1
+        end
+      end
+    end
+
+    describe '#count_failed' do
+      let!(:job_artifact_1) { create(:ci_job_artifact, project: synced_project) }
+      let!(:job_artifact_2) { create(:ci_job_artifact, project: unsynced_project) }
+      let!(:job_artifact_3) { create(:ci_job_artifact, project: synced_project) }
+      let!(:job_artifact_4) { create(:ci_job_artifact, project: unsynced_project) }
+      let!(:job_artifact_5) { create(:ci_job_artifact, project: project_broken_storage) }
+      let!(:job_artifact_6) { create(:ci_job_artifact, project: project_broken_storage) }
+
+      context 'without selective sync' do
+        before do
+          create(:geo_job_artifact_registry, artifact_id: job_artifact_1.id, success: false)
+          create(:geo_job_artifact_registry, artifact_id: job_artifact_2.id)
+          create(:geo_job_artifact_registry, artifact_id: job_artifact_3.id, success: false)
+          create(:geo_job_artifact_registry, artifact_id: job_artifact_5.id, success: false)
+        end
+
+        it 'counts job artifacts that sync has failed' do
+          expect(subject.count_failed).to eq 3
+        end
+
+        it 'ignores remote job artifacts' do
+          job_artifact_1.update_column(:file_store, ObjectStorage::Store::REMOTE)
+
+          expect(subject.count_failed).to eq 2
+        end
+
+        it 'ignores expired job artifacts' do
+          job_artifact_1.update_column(:expire_at, Date.yesterday)
+
+          expect(subject.count_failed).to eq 2
+        end
+      end
+
+      context 'with selective sync by namespace' do
+        before do
+          secondary.update!(selective_sync_type: 'namespaces', namespaces: [synced_group])
+
+          create(:geo_job_artifact_registry, artifact_id: job_artifact_1.id, success: false)
+          create(:geo_job_artifact_registry, artifact_id: job_artifact_2.id, success: false)
+          create(:geo_job_artifact_registry, artifact_id: job_artifact_3.id, success: false)
+          create(:geo_job_artifact_registry, artifact_id: job_artifact_5.id, success: false)
+        end
+
+        it 'counts job artifacts that sync has failed' do
+          expect(subject.count_failed).to eq 2
+        end
+
+        it 'ignores remote job artifacts' do
+          job_artifact_1.update_column(:file_store, ObjectStorage::Store::REMOTE)
+
+          expect(subject.count_failed).to eq 1
+        end
+
+        it 'ignores expired job artifacts' do
+          job_artifact_1.update_column(:expire_at, Date.yesterday)
+
+          expect(subject.count_failed).to eq 1
+        end
+      end
+
+      context 'with selective sync by shard' do
+        before do
+          secondary.update!(selective_sync_type: 'shards', selective_sync_shards: ['broken'])
+
+          create(:geo_job_artifact_registry, artifact_id: job_artifact_1.id, success: false)
+          create(:geo_job_artifact_registry, artifact_id: job_artifact_2.id, success: false)
+          create(:geo_job_artifact_registry, artifact_id: job_artifact_3.id, success: false)
+          create(:geo_job_artifact_registry, artifact_id: job_artifact_5.id, success: false)
+          create(:geo_job_artifact_registry, artifact_id: job_artifact_6.id, success: false)
+        end
+
+        it 'counts job artifacts that sync has failed' do
+          expect(subject.count_failed).to eq 2
+        end
+
+        it 'ignores remote job artifacts' do
+          job_artifact_5.update_column(:file_store, ObjectStorage::Store::REMOTE)
+
+          expect(subject.count_failed).to eq 1
+        end
+
+        it 'ignores expired job artifacts' do
+          job_artifact_5.update_column(:expire_at, Date.yesterday)
+
+          expect(subject.count_failed).to eq 1
+        end
+      end
+    end
+
+    describe '#count_synced_missing_on_primary' do
+      let!(:job_artifact_1) { create(:ci_job_artifact, project: synced_project) }
+      let!(:job_artifact_2) { create(:ci_job_artifact, project: unsynced_project) }
+      let!(:job_artifact_3) { create(:ci_job_artifact, project: synced_project) }
+      let!(:job_artifact_4) { create(:ci_job_artifact, project: unsynced_project) }
+      let!(:job_artifact_5) { create(:ci_job_artifact, project: project_broken_storage) }
+      let!(:job_artifact_6) { create(:ci_job_artifact, project: project_broken_storage) }
+
+      context 'without selective sync' do
+        before do
+          create(:geo_job_artifact_registry, artifact_id: job_artifact_1.id, success: false, missing_on_primary: false)
+          create(:geo_job_artifact_registry, artifact_id: job_artifact_2.id)
+          create(:geo_job_artifact_registry, artifact_id: job_artifact_3.id, missing_on_primary: true)
+          create(:geo_job_artifact_registry, artifact_id: job_artifact_4.id, missing_on_primary: false)
+          create(:geo_job_artifact_registry, artifact_id: job_artifact_5.id, missing_on_primary: true)
+          create(:geo_job_artifact_registry, artifact_id: job_artifact_6.id)
+        end
+
+        it 'counts job artifacts that have been synced and are missing on the primary' do
+          expect(subject.count_synced_missing_on_primary).to eq 2
+        end
+
+        it 'ignores remote job artifacts' do
+          job_artifact_3.update_column(:file_store, ObjectStorage::Store::REMOTE)
+
+          expect(subject.count_synced_missing_on_primary).to eq 1
+        end
+
+        it 'ignores expired job artifacts' do
+          job_artifact_3.update_column(:expire_at, Date.yesterday)
+
+          expect(subject.count_synced_missing_on_primary).to eq 1
+        end
+      end
+
+      context 'with selective sync by namespace' do
+        before do
+          secondary.update!(selective_sync_type: 'namespaces', namespaces: [synced_group])
+
+          create(:geo_job_artifact_registry, artifact_id: job_artifact_1.id, missing_on_primary: true)
+          create(:geo_job_artifact_registry, artifact_id: job_artifact_2.id)
+          create(:geo_job_artifact_registry, artifact_id: job_artifact_3.id, missing_on_primary: true)
+          create(:geo_job_artifact_registry, artifact_id: job_artifact_4.id, missing_on_primary: false)
+          create(:geo_job_artifact_registry, artifact_id: job_artifact_5.id, missing_on_primary: true)
+          create(:geo_job_artifact_registry, artifact_id: job_artifact_6.id)
+        end
+
+        it 'counts job artifacts that have been synced and are missing on the primary' do
+          expect(subject.count_synced_missing_on_primary).to eq 2
+        end
+
+        it 'ignores remote job artifacts' do
+          job_artifact_1.update_column(:file_store, ObjectStorage::Store::REMOTE)
+
+          expect(subject.count_synced_missing_on_primary).to eq 1
+        end
+
+        it 'ignores expired job artifacts' do
+          job_artifact_1.update_column(:expire_at, Date.yesterday)
+
+          expect(subject.count_synced_missing_on_primary).to eq 1
+        end
+      end
+
+      context 'with selective sync by shard' do
+        before do
+          secondary.update!(selective_sync_type: 'shards', selective_sync_shards: ['broken'])
+
+          create(:geo_job_artifact_registry, artifact_id: job_artifact_1.id, missing_on_primary: true)
+          create(:geo_job_artifact_registry, artifact_id: job_artifact_2.id)
+          create(:geo_job_artifact_registry, artifact_id: job_artifact_3.id, missing_on_primary: true)
+          create(:geo_job_artifact_registry, artifact_id: job_artifact_4.id, missing_on_primary: false)
+          create(:geo_job_artifact_registry, artifact_id: job_artifact_5.id, missing_on_primary: true)
+          create(:geo_job_artifact_registry, artifact_id: job_artifact_6.id, missing_on_primary: true)
+        end
+
+        it 'counts job artifacts that have been synced and are missing on the primary' do
+          expect(subject.count_synced_missing_on_primary).to eq 2
+        end
+
+        it 'ignores remote job artifacts' do
+          job_artifact_5.update_column(:file_store, ObjectStorage::Store::REMOTE)
+
+          expect(subject.count_synced_missing_on_primary).to eq 1
+        end
+
+        it 'ignores expired job artifacts' do
+          job_artifact_5.update_column(:expire_at, Date.yesterday)
+
+          expect(subject.count_synced_missing_on_primary).to eq 1
+        end
+      end
+    end
+
+    describe '#count_registry' do
+      let!(:job_artifact_1) { create(:ci_job_artifact, project: synced_project) }
+      let!(:job_artifact_2) { create(:ci_job_artifact, project: unsynced_project) }
+      let!(:job_artifact_3) { create(:ci_job_artifact, project: synced_project) }
+      let!(:job_artifact_4) { create(:ci_job_artifact, project: unsynced_project) }
+      let!(:job_artifact_5) { create(:ci_job_artifact, project: project_broken_storage) }
+      let!(:job_artifact_6) { create(:ci_job_artifact, project: project_broken_storage) }
+
+      before do
+        create(:geo_job_artifact_registry, artifact_id: job_artifact_1.id, success: false)
+        create(:geo_job_artifact_registry, artifact_id: job_artifact_3.id, missing_on_primary: true)
+        create(:geo_job_artifact_registry, artifact_id: job_artifact_4.id)
+        create(:geo_job_artifact_registry, artifact_id: job_artifact_6.id)
+      end
+
+      it 'counts file registries for job artifacts' do
+        expect(subject.count_registry).to eq 4
+      end
+
+      context 'with selective sync by namespace' do
+        before do
+          secondary.update!(selective_sync_type: 'namespaces', namespaces: [synced_group])
+        end
+
+        it 'does not apply the selective sync restriction' do
+          expect(subject.count_registry).to eq 4
+        end
+      end
+
+      context 'with selective sync by shard' do
+        before do
+          secondary.update!(selective_sync_type: 'shards', selective_sync_shards: ['broken'])
+        end
+
+        it 'does not apply the selective sync restriction' do
+          expect(subject.count_registry).to eq 4
+        end
+      end
+    end
+
+    include_examples 'finds all the things'
+  end
 end