Allow locking refs whilst filtering for fetch requests

Allow the lock/unlock pattern when filtering the refs to fetch. This
mechanism is used for making sure that the refs selected for the fetch
replication operation do not move throughout the replication process.

This is needed for preventing a ref-update operation happening
concurrently with the replication process invalidating the filtering
validity.

Change-Id: I6a30aff43a81f00d671554186eb6e796446b148d

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

diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/FetchOne.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/FetchOne.java
index 939a05b..a5c66c0 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/FetchOne.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/FetchOne.java
@@ -109,6 +109,7 @@
   private final Optional<PullReplicationApiRequestMetrics> apiRequestMetrics;
   private DynamicItem<ReplicationFetchFilter> replicationFetchFilter;
   private boolean succeeded;
+  private Map<String, AutoCloseable> fetchLocks;
 
   @Inject
   FetchOne(
@@ -507,6 +508,8 @@
     } catch (IOException e) {
       notifyRefReplicatedIOException();
       throw e;
+    } finally {
+      unlockRefSpecs(fetchLocks);
     }
     return fetchRefSpecs;
   }
@@ -556,6 +559,25 @@
         .collect(Collectors.toList());
   }
 
+  public void unlockRefSpecs(Map<String, AutoCloseable> locks) {
+    if (locks == null) {
+      return;
+    }
+
+    locks.forEach(
+        (key, value) -> {
+          try {
+            value.close();
+          } catch (Exception e) {
+            repLog.error(
+                String.format(
+                    "[%s] Error when unlocking ref %s:%s", taskIdHex, projectName.get(), key),
+                e);
+          }
+        });
+    locks.clear();
+  }
+
   public List<FetchRefSpec> safeGetFetchRefSpecs() {
     try {
       return getFetchRefSpecs();
@@ -610,7 +632,11 @@
         refs.stream().map(FetchRefSpec::refName).collect(Collectors.toUnmodifiableSet());
     Set<String> filteredRefNames =
         replicationFetchFilter()
-            .map(f -> f.filter(this.projectName.get(), refsNames))
+            .map(
+                f -> {
+                  fetchLocks = f.filterAndLock(this.projectName.get(), refsNames);
+                  return fetchLocks.keySet();
+                })
             .orElse(refsNames);
     return refs.stream()
         .filter(refSpec -> filteredRefNames.contains(refSpec.refName()))

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

diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/ReplicationFetchFilter.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/ReplicationFetchFilter.java
index 77f4185..17afe94 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/ReplicationFetchFilter.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/ReplicationFetchFilter.java
@@ -15,7 +15,9 @@
 package com.googlesource.gerrit.plugins.replication.pull;
 
 import com.google.gerrit.extensions.annotations.ExtensionPoint;
+import java.util.Map;
 import java.util.Set;
+import java.util.stream.Collectors;
 
 /**
  * Filter that is invoked before a set of remote refs are fetched from a remote instance.
@@ -26,4 +28,9 @@
 public interface ReplicationFetchFilter {
 
   Set<String> filter(String projectName, Set<String> fetchRefs);
+
+  default Map<String, AutoCloseable> filterAndLock(String projectName, Set<String> fetchRefs) {
+    return filter(projectName, fetchRefs).stream()
+        .collect(Collectors.toMap(ref -> ref, ref -> () -> {}));
+  }
 }