Distributor: Skip re-fire of already pending tasks On every cycle the distributor streams all tasks under waiting/ and calls fireFromStorage() for each one, which allocates a fresh ReplicationState and consolidates it into the pending PushOne for that URI. PushOne deduplicates refs, since refBatchesToPush is a Set, but its stateMap is a ListMultimap that appends unconditionally. A task that stays in waiting/ across cycles therefore grows the pending task's notification list by one entry per ref per cycle while adding no work at all. The distributor already snapshots the ref updates carried by pending push tasks, in order to prune queue entries whose stored counterpart is gone. Each streamed update is removed from that snapshot, leaving only the prunable ones behind by the end. Reuse that removal as the test, i.e a non-null result means a pending task on this node is already holding the update, so re-firing it would append a redundant state. Skip it, and let only updates with no pending task reach fireFromStorage(). Tasks written by another primary are absent from the snapshot and continue to be fired, so cluster distribution is unaffected. This change removes the repeat re-fires when Prune.TRUE rather than every possible source of stateMap growth. The startup replay passes Prune.FALSE and is deliberately left uncovered. It walks each waiting/ file exactly once, so no task is fired twice into the same pending push and the growth is not reachable there today. Change-Id: Ib9e77f9cf99cab3d5a6f110f0578ae02eafe9198
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationQueue.java b/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationQueue.java index 9abf407..9827091 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationQueue.java +++ b/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationQueue.java
@@ -269,10 +269,13 @@ @Override public void run(ReplicationTasksStorage.ReplicateRefUpdate u) { try { - fireFromStorage(new URIish(u.uri()), Project.nameKey(u.project()), u.refs()); if (Prune.TRUE.equals(prune)) { - taskNamesByReplicateRefUpdate.remove(u); + if (taskNamesByReplicateRefUpdate.remove(u) != null) { + repLog.atFine().log("Task %s is already scheduled, not re-firing", u); + return; + } } + fireFromStorage(new URIish(u.uri()), Project.nameKey(u.project()), u.refs()); } catch (URISyntaxException e) { repLog.atSevere().withCause(e).log( "Encountered malformed URI for persisted event %s", u);
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationDistributorIT.java b/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationDistributorIT.java index dc73036..5256bbe 100644 --- a/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationDistributorIT.java +++ b/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationDistributorIT.java
@@ -15,6 +15,7 @@ package com.googlesource.gerrit.plugins.replication; import static com.google.common.truth.Truth.assertThat; +import static com.google.gerrit.testing.GerritJUnit.assertThrows; import com.google.gerrit.acceptance.TestPlugin; import com.google.gerrit.acceptance.UseLocalDisk; @@ -22,6 +23,7 @@ import com.google.gerrit.entities.BranchNameKey; import com.google.gerrit.entities.Project; import com.google.gerrit.server.git.WorkQueue; +import com.googlesource.gerrit.plugins.replication.api.ReplicationConfig.FilterType; import java.time.Duration; import java.util.List; import java.util.Set; @@ -85,6 +87,40 @@ } @Test + public void distributorDoesNotReFirePendingTask() throws Exception { + String remote = "foo"; + String replica = "replica"; + String master = "refs/heads/master"; + String pendingBranch = "refs/heads/pending_branch"; + String otherPrimaryBranch = "refs/heads/other_primary_branch"; + Project.NameKey targetProject = createTestProject(project + replica); + URIish targetUri = new URIish(getProjectUri(targetProject)); + setReplicationDestination(remote, replica, ALL_PROJECTS, TEST_LONG_REPLICATION_DELAY_SECONDS); + reloadConfig(); + + createBranch(BranchNameKey.create(project, pendingBranch)); + assertThat(listWaitingReplicationTasks(pendingBranch)).hasSize(1); + PushOne pendingPush = getPendingPush(remote, targetUri); + assertThat(pendingPush.getStatesByRef(pendingBranch)).hasLength(1); + + createBranch(project, master, otherPrimaryBranch); + tasksStorage.create( + ReplicationTasksStorage.ReplicateRefUpdate.create( + project.get(), Set.of(otherPrimaryBranch), targetUri, remote)); + + WaitUtil.waitUntil( + () -> pendingPush.getStatesByRef(otherPrimaryBranch).length == 1, + Duration.ofSeconds(TEST_DISTRIBUTION_CYCLE_SECONDS)); + + assertThrows( + InterruptedException.class, + () -> + WaitUtil.waitUntil( + () -> pendingPush.getStatesByRef(pendingBranch).length > 1, + Duration.ofSeconds(TEST_DISTRIBUTION_CYCLE_SECONDS))); + } + + @Test public void distributorPrunesTaskFromWorkQueue() throws Exception { createTestProject(project + "replica"); setReplicationDestination("foo", "replica", ALL_PROJECTS, Integer.MAX_VALUE); @@ -100,6 +136,16 @@ .isTrue(); } + private PushOne getPendingPush(String remote, URIish uri) { + return destinationCollection.getAll(FilterType.ALL).stream() + .filter(dest -> remote.equals(dest.getRemoteConfigName())) + .findFirst() + .get() + .getQueue() + .pending + .get(uri); + } + private List<WorkQueue.Task<?>> getProjectTasks() { return getInstance(WorkQueue.class).getTasks().stream() .filter(t -> t instanceof WorkQueue.ProjectTask)
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationQueueTest.java b/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationQueueTest.java index 0fabf57..adb16c9 100644 --- a/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationQueueTest.java +++ b/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationQueueTest.java
@@ -99,6 +99,16 @@ } @Test + public void distributorDoesNotFireTaskPendingOnThisNode() throws Exception { + start(); + waitingTasks.add(update); + taskNamesByReplicateRefUpdate.put(update, "pending push task"); + runDistributor(); + + verify(destination, never()).scheduleFromStorage(any(), any(), any(), any()); + } + + @Test public void distributorFiresTaskNotPendingOnThisNode() throws Exception { start(); waitingTasks.add(update);