Implement filterAndLock in ReplicationFetchFilter The ReplicationFetchFilter allows to also lock refs whilst filtering them, which is paramount for preventing the refs to be fetch from being mutated and therefore causing potential split-brains. Lock local refs locally for allowing concurrent access to happen on other nodes that are not impacted by the local replication tasks. Change-Id: I0f8778dee71b3e3963403541b212058cdd42c784
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/Configuration.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/Configuration.java index a7a05d3..4415aea 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/multisite/Configuration.java +++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/Configuration.java
@@ -58,6 +58,8 @@ private static final String REPLICATION_LAG_REFRESH_INTERVAL = "replicationLagRefreshInterval"; private static final String REPLICATION_LAG_ENABLED = "replicationLagEnabled"; private static final Duration REPLICATION_LAG_REFRESH_INTERVAL_DEFAULT = Duration.ofSeconds(60); + private static final String LOCAL_REF_LOCK_TIMEOUT = "localRefLockTimeout"; + private static final Duration LOCAL_REF_LOCK_TIMEOUT_DEFAULT = Duration.ofSeconds(30); private static final String REPLICATION_CONFIG = "replication.config"; // common parameters to cache and index sections @@ -79,6 +81,7 @@ private final Config multiSiteConfig; private final Supplier<Duration> replicationLagRefreshInterval; private final Supplier<Boolean> replicationLagEnabled; + private final Supplier<Long> localRefLockTimeoutMsec; @Inject Configuration(SitePaths sitePaths) { @@ -118,6 +121,16 @@ lazyMultiSiteCfg .get() .getBoolean(REF_DATABASE, null, REPLICATION_LAG_ENABLED, true)); + localRefLockTimeoutMsec = + memoize( + () -> + ConfigUtil.getTimeUnit( + lazyMultiSiteCfg.get(), + REF_DATABASE, + null, + LOCAL_REF_LOCK_TIMEOUT, + LOCAL_REF_LOCK_TIMEOUT_DEFAULT.toMillis(), + TimeUnit.MILLISECONDS)); } public Config getMultiSiteConfig() { @@ -160,6 +173,10 @@ return replicationLagEnabled.get(); } + public long localRefLockTimeoutMsec() { + return localRefLockTimeoutMsec.get(); + } + public Collection<Message> validate() { return replicationConfigValidation.get(); }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/validation/MultisiteReplicationFetchFilter.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/validation/MultisiteReplicationFetchFilter.java index d67492d..ca79bcd 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/multisite/validation/MultisiteReplicationFetchFilter.java +++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/validation/MultisiteReplicationFetchFilter.java
@@ -17,7 +17,9 @@ import static com.googlesource.gerrit.plugins.replication.pull.PullReplicationLogger.repLog; import com.gerritforge.gerrit.globalrefdb.GlobalRefDbLockException; +import com.gerritforge.gerrit.globalrefdb.RefDbLockException; import com.gerritforge.gerrit.globalrefdb.validation.SharedRefDatabaseWrapper; +import com.google.common.collect.Sets; import com.google.common.flogger.FluentLogger; import com.google.gerrit.entities.Project; import com.google.gerrit.server.git.GitRepositoryManager; @@ -27,6 +29,9 @@ import com.googlesource.gerrit.plugins.replication.pull.ReplicationFetchFilter; import java.io.IOException; import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Map; import java.util.Optional; import java.util.Set; import java.util.stream.Collectors; @@ -89,6 +94,36 @@ } } + @Override + public Map<String, AutoCloseable> filterAndLock(String projectName, Set<String> fetchRefs) { + Project.NameKey projectKey = Project.nameKey(projectName); + Set<String> filteredRefs = new HashSet<>(); + Map<String, AutoCloseable> refLocks = new HashMap<>(); + try { + for (String ref : fetchRefs) { + refLocks.put(ref, sharedRefDb.lockLocalRef(projectKey, ref)); + } + filteredRefs.addAll(filter(projectName, fetchRefs)); + } catch (RefDbLockException lockException) { + filteredRefs.clear(); + throw lockException; + } finally { + for (String excludedRef : Sets.difference(fetchRefs, filteredRefs)) { + AutoCloseable excludedLock = refLocks.remove(excludedRef); + if (excludedLock != null) { + try { + excludedLock.close(); + } catch (Exception e) { + logger.atWarning().withCause(e).log( + "Error whilst unlocking ref %s:%s", projectName, excludedRef); + } + } + } + } + + return refLocks; + } + private Optional<ObjectId> getLocalSha1IfEqualsToExistingGlobalRefDb( Repository repository, String projectName,
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/validation/ReentrantRefDbLocker.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/validation/ReentrantRefDbLocker.java new file mode 100644 index 0000000..b2085ae --- /dev/null +++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/validation/ReentrantRefDbLocker.java
@@ -0,0 +1,74 @@ +// Copyright (C) 2025 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.multisite.validation; + +import com.gerritforge.gerrit.globalrefdb.RefDbLockException; +import com.gerritforge.gerrit.globalrefdb.validation.RefLocker; +import com.google.gerrit.entities.Project; +import com.google.inject.Inject; +import com.googlesource.gerrit.plugins.multisite.Configuration; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; + +class ReentrantRefDbLocker implements RefLocker { + private final ConcurrentHashMap<String, RefLock> refsLocks; + private final long timeoutMsec; + + private class RefLock implements AutoCloseable { + private final Lock lock; + + RefLock() { + lock = new ReentrantLock(); + } + + boolean lock() throws InterruptedException { + return lock.tryLock(timeoutMsec, TimeUnit.MILLISECONDS); + } + + @Override + public void close() throws Exception { + lock.unlock(); + } + } + + @Inject + public ReentrantRefDbLocker(Configuration configuration) { + this.timeoutMsec = configuration.localRefLockTimeoutMsec(); + refsLocks = new ConcurrentHashMap<>(); + } + + @Override + public AutoCloseable lockRef(Project.NameKey project, String refName) throws RefDbLockException { + RefLock lock = refsLocks.computeIfAbsent(getKey(project, refName), ref -> new RefLock()); + try { + if (lock.lock()) { + return lock; + } + + throw new RefDbLockException( + project.get(), + refName, + String.format("Unable to acquire local ref lock after %s msec", timeoutMsec)); + } catch (InterruptedException e) { + throw new RefDbLockException(project.get(), refName, e); + } + } + + private String getKey(Project.NameKey project, String refName) { + return String.format("%s:%s", project.get(), refName); + } +}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/validation/ValidationModule.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/validation/ValidationModule.java index c7e2a9a..6f8ec50 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/multisite/validation/ValidationModule.java +++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/validation/ValidationModule.java
@@ -16,6 +16,7 @@ import com.gerritforge.gerrit.globalrefdb.validation.BatchRefUpdateValidator; import com.gerritforge.gerrit.globalrefdb.validation.Log4jSharedRefLogger; +import com.gerritforge.gerrit.globalrefdb.validation.RefLocker; import com.gerritforge.gerrit.globalrefdb.validation.RefUpdateValidator; import com.gerritforge.gerrit.globalrefdb.validation.SharedRefDatabaseWrapper; import com.gerritforge.gerrit.globalrefdb.validation.SharedRefDbBatchRefUpdate; @@ -46,6 +47,7 @@ @Override protected void configure() { + bind(RefLocker.class).to(ReentrantRefDbLocker.class).in(Scopes.SINGLETON); bind(SharedRefDatabaseWrapper.class).in(Scopes.SINGLETON); bind(SharedRefLogger.class).to(Log4jSharedRefLogger.class);
diff --git a/src/main/resources/Documentation/config.md b/src/main/resources/Documentation/config.md index 7547fd9..55e5041 100644 --- a/src/main/resources/Documentation/config.md +++ b/src/main/resources/Documentation/config.md
@@ -95,6 +95,11 @@ : Enable the use of a shared ref-database Defaults: true +```ref-database.localRefLockTimeout``` +: Timeout waiting for a local ref to become available to accept + updates or for starting a replication task. + Defaults: 30 sec + ```ref-database.replicationLagEnabled``` : Enable the metrics to trace the auto-replication lag between sites updating the `refs/multi-site/version/*` to the _epoch_ timestamp in