Ensure that replicateOnStartup doesn't compete with fetch

When `gerrit.replicateOnStartup` is set to `true` then corresponding
projects will be fetched. Ensure that projects that are subject of
periodic fetch are excluded from the `replicateOnStartup` schedule as
they are going to be fetched anyway.

Bug: Issue 322146240
Change-Id: I29c6b12fc12ba0860f43a7fb2d116faa2482a1fa

src/main/java/com/googlesource/gerrit/plugins/replication/pull/OnStartStop.java

diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/OnStartStop.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/OnStartStop.java
index 0a75c86..6ac09f6 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/OnStartStop.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/OnStartStop.java
@@ -25,7 +25,6 @@
 import com.google.inject.Inject;
 import com.google.inject.Provider;
 import com.googlesource.gerrit.plugins.replication.ReplicationConfig;
-import com.googlesource.gerrit.plugins.replication.ReplicationFilter;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
@@ -79,7 +78,11 @@
               new FetchResultProcessing.GitUpdateProcessing(eventDispatcher.get()));
       fetchAllFuture.set(
           fetchAll
-              .create(null, ReplicationFilter.all(), state, false)
+              .create(
+                  null,
+                  fetchAllPeriodically.get().skipFromReplicateAllOnPluginStart(),
+                  state,
+                  false)
               .schedule(30, TimeUnit.SECONDS));
     }
   }

src/main/java/com/googlesource/gerrit/plugins/replication/pull/SourceFetchPeriodically.java

diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/SourceFetchPeriodically.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/SourceFetchPeriodically.java
index 2c2f6d8..264c880 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/SourceFetchPeriodically.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/SourceFetchPeriodically.java
@@ -15,6 +15,7 @@
 package com.googlesource.gerrit.plugins.replication.pull;
 
 import com.google.common.flogger.FluentLogger;
+import com.google.gerrit.entities.Project;
 import com.google.gerrit.extensions.registration.DynamicItem;
 import com.google.gerrit.server.events.EventDispatcher;
 import com.google.gerrit.server.project.ProjectCache;
@@ -26,6 +27,7 @@
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
+import java.util.stream.Stream;
 
 class SourceFetchPeriodically {
   interface Factory {
@@ -59,11 +61,14 @@
         this::scheduleFetchAll, 0L, source.fetchEvery(), TimeUnit.SECONDS);
   }
 
+  Stream<Project.NameKey> projectsToFetch() {
+    return projects.all().stream().filter(source::wouldFetchProject);
+  }
+
   private void scheduleFetchAll() {
     Optional<PullReplicationApiRequestMetrics> metrics = Optional.of(metricsProvider.get());
     long repositoriesToBeFetched =
-        projects.all().stream()
-            .filter(source::wouldFetchProject)
+        projectsToFetch()
             .map(
                 projectToFetch ->
                     source.scheduleNow(

src/main/java/com/googlesource/gerrit/plugins/replication/pull/SourcesFetchPeriodically.java

diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/SourcesFetchPeriodically.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/SourcesFetchPeriodically.java
index b405a6e..de1d571 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/SourcesFetchPeriodically.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/SourcesFetchPeriodically.java
@@ -16,15 +16,20 @@
 
 import static com.googlesource.gerrit.plugins.replication.pull.SourceConfiguration.DEFAULT_PERIODIC_FETCH_DISABLED;
 import static java.util.stream.Collectors.toList;
+import static java.util.stream.Collectors.toSet;
 
 import com.google.common.base.Suppliers;
 import com.google.common.flogger.FluentLogger;
+import com.google.gerrit.entities.Project;
 import com.google.gerrit.server.git.WorkQueue;
 import com.google.inject.Inject;
 import com.google.inject.Provider;
 import com.google.inject.Singleton;
+import com.googlesource.gerrit.plugins.replication.ReplicationFilter;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
+import java.util.Set;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
 import java.util.function.Supplier;
@@ -36,7 +41,7 @@
   private final WorkQueue workQueue;
   private final Provider<SourcesCollection> sources;
   private final Provider<SourceFetchPeriodically.Factory> fetchAllCreator;
-  private final List<ScheduledFuture<?>> scheduled;
+  private final List<ScheduledSource> scheduled;
 
   @Inject
   SourcesFetchPeriodically(
@@ -53,7 +58,19 @@
     scheduled.addAll(scheduleFetchAll(workQueue, sources.get(), fetchAllCreator.get()));
   }
 
-  private List<ScheduledFuture<?>> scheduleFetchAll(
+  ReplicationFilter skipFromReplicateAllOnPluginStart() {
+    Set<Project.NameKey> projectsToSkip =
+        scheduled.stream()
+            .flatMap(scheduledSource -> scheduledSource.sourceFetch.projectsToFetch())
+            .collect(toSet());
+    if (projectsToSkip.isEmpty()) {
+      return ReplicationFilter.all();
+    }
+
+    return new SkipProjectsFilter(projectsToSkip);
+  }
+
+  private List<ScheduledSource> scheduleFetchAll(
       WorkQueue workQueue,
       SourcesCollection sources,
       SourceFetchPeriodically.Factory fetchAllCreator) {
@@ -66,12 +83,41 @@
               logger.atInfo().log(
                   "Enabling periodic (every %ds) fetch of all refs for [%s] remote",
                   source.fetchEvery(), source.getRemoteConfigName());
-              return fetchAllCreator.create(source).start(queue.get());
+              SourceFetchPeriodically sourceFetchAll = fetchAllCreator.create(source);
+              return new ScheduledSource(sourceFetchAll, sourceFetchAll.start(queue.get()));
             })
         .collect(toList());
   }
 
   void stop() {
-    scheduled.forEach(schedule -> schedule.cancel(true));
+    scheduled.forEach(source -> source.schedule.cancel(true));
+  }
+
+  private static class ScheduledSource {
+    private final SourceFetchPeriodically sourceFetch;
+    private final ScheduledFuture<?> schedule;
+
+    private ScheduledSource(SourceFetchPeriodically sourceFetch, ScheduledFuture<?> schedule) {
+      this.sourceFetch = sourceFetch;
+      this.schedule = schedule;
+    }
+  }
+
+  /**
+   * The ReplicationFilter implementation that matches all projects that are not part of the
+   * internal set.
+   */
+  private static class SkipProjectsFilter extends ReplicationFilter {
+    private final Set<Project.NameKey> projectsToSkip;
+
+    private SkipProjectsFilter(Set<Project.NameKey> projectsToSkip) {
+      super(Collections.emptyList());
+      this.projectsToSkip = projectsToSkip;
+    }
+
+    @Override
+    public boolean matches(Project.NameKey name) {
+      return !projectsToSkip.contains(name);
+    }
   }
 }

src/test/java/com/googlesource/gerrit/plugins/replication/pull/SourcesFetchPeriodicallyTest.java

diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/pull/SourcesFetchPeriodicallyTest.java b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/SourcesFetchPeriodicallyTest.java
new file mode 100644
index 0000000..47b891d
--- /dev/null
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/SourcesFetchPeriodicallyTest.java
@@ -0,0 +1,86 @@
+// Copyright (C) 2024 The Android Open Source Project
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.googlesource.gerrit.plugins.replication.pull;
+
+import static com.google.common.truth.Truth.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import com.google.gerrit.entities.Project;
+import com.google.gerrit.entities.Project.NameKey;
+import com.google.gerrit.server.git.WorkQueue;
+import com.google.inject.util.Providers;
+import com.googlesource.gerrit.plugins.replication.ReplicationFilter;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Stream;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+
+@RunWith(MockitoJUnitRunner.class)
+public class SourcesFetchPeriodicallyTest {
+  private static final NameKey RANDOM_PROJECT_NAME = Project.nameKey("random_project_name");
+
+  @Mock WorkQueue workQueueMock;
+  @Mock SourcesCollection sourcesMock;
+  @Mock SourceFetchPeriodically.Factory factoryMock;
+
+  private SourcesFetchPeriodically objectUnderTest;
+
+  @Before
+  public void setup() {
+    objectUnderTest =
+        new SourcesFetchPeriodically(
+            workQueueMock, Providers.of(sourcesMock), Providers.of(factoryMock));
+  }
+
+  @Test
+  public void shouldMatchAnyProjectWhenNoProjectsToFetchPeriodicallyAreConfigured() {
+    // given
+    when(sourcesMock.getAll()).thenReturn(Collections.emptyList());
+
+    // when
+    objectUnderTest.start();
+    ReplicationFilter filter = objectUnderTest.skipFromReplicateAllOnPluginStart();
+
+    // then
+    assertThat(filter.matches(RANDOM_PROJECT_NAME)).isTrue();
+  }
+
+  @Test
+  public void shouldNotMatchProjectConfiguredToFetchPeriodically() {
+    // given
+    Project.NameKey projectToFetch = Project.nameKey("to_be_fetched");
+    SourceFetchPeriodically fetchProjectForSource = mock(SourceFetchPeriodically.class);
+    when(fetchProjectForSource.projectsToFetch()).thenReturn(Stream.of(projectToFetch));
+
+    Source sourceWithPeriodicFetch = mock(Source.class);
+    when(sourceWithPeriodicFetch.fetchEvery()).thenReturn(1L);
+    when(sourcesMock.getAll()).thenReturn(List.of(sourceWithPeriodicFetch));
+    when(factoryMock.create(any())).thenReturn(fetchProjectForSource);
+
+    // when
+    objectUnderTest.start();
+    ReplicationFilter filter = objectUnderTest.skipFromReplicateAllOnPluginStart();
+
+    // then
+    assertThat(filter.matches(projectToFetch)).isFalse();
+    assertThat(filter.matches(RANDOM_PROJECT_NAME)).isTrue();
+  }
+}