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 d60c2b7..76ecd80 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
@@ -113,6 +113,7 @@
   private DynamicItem<ReplicationFetchFilter> replicationFetchFilter;
   private boolean succeeded;
   private Map<String, AutoCloseable> fetchLocks;
+  private final ProjectsLock projectsLock;
 
   @Inject
   FetchOne(
@@ -126,6 +127,7 @@
       FetchFactory fetchFactory,
       FetchRefsDatabase fetchRefsDatabase,
       DeleteRefCommand deleteRefCommand,
+      ProjectsLock projectsLock,
       @Assisted Project.NameKey d,
       @Assisted URIish u,
       @Assisted Optional<PullReplicationApiRequestMetrics> apiRequestMetrics) {
@@ -148,6 +150,7 @@
     this.fetchFactory = fetchFactory;
     maxRetries = s.getMaxRetries();
     this.apiRequestMetrics = apiRequestMetrics;
+    this.projectsLock = projectsLock;
   }
 
   @Inject(optional = true)
@@ -350,126 +353,142 @@
     // created and scheduled for a future point in time.)
     //
     isCollision = false;
-    if (replicationType == ReplicationType.ASYNC && !pool.requestRunway(this)) {
-      if (!canceled) {
-        repLog.info(
-            "[{}] Rescheduling replication from {} to avoid collision with an in-flight fetch task"
-                + " [{}].",
-            taskIdHex,
-            uri,
-            pool.getInFlight(getURI()).map(FetchOne::getTaskIdHex).orElse("<unknown>"));
-        pool.reschedule(this, Source.RetryReason.COLLISION);
-        isCollision = true;
+
+    try (ProjectsLock.LockToken unused = projectsLock.tryLock(projectName, getTaskIdHex())) {
+      if (replicationType == ReplicationType.ASYNC && !pool.requestRunway(this)) {
+        if (!canceled) {
+          repLog.info(
+              "[{}] Rescheduling replication from {} to avoid collision with an in-flight fetch"
+                  + " task [{}].",
+              taskIdHex,
+              uri,
+              pool.getInFlight(getURI()).map(FetchOne::getTaskIdHex).orElse("<unknown>"));
+          pool.reschedule(this, Source.RetryReason.COLLISION);
+          isCollision = true;
+        }
+        return;
       }
-      return;
-    }
 
-    repLog.info(
-        "[{}] {} replication from {} started for refs [{}] ...",
-        taskIdHex,
-        replicationType,
-        uri,
-        getRefSpecs());
-    Timer1.Context<String> context = metrics.start(config.getName());
-    try {
-      long startedAt = context.getStartTime();
-      long delay = NANOSECONDS.toMillis(startedAt - createdAt);
-      git = gitManager.openRepository(projectName);
-      List<FetchRefSpec> fetchRefSpecs = runImpl();
+      repLog.info(
+          "[{}] {} replication from {} started for refs [{}] ...",
+          taskIdHex,
+          replicationType,
+          uri,
+          getRefSpecs());
+      Timer1.Context<String> context = metrics.start(config.getName());
+      try {
+        long startedAt = context.getStartTime();
+        long delay = NANOSECONDS.toMillis(startedAt - createdAt);
+        git = gitManager.openRepository(projectName);
+        List<FetchRefSpec> fetchRefSpecs = runImpl();
 
-      if (fetchRefSpecs.isEmpty()) {
-        repLog.info(
-            "[{}] {} replication from {} finished but no refs were replicated, {}ms delay, {}"
-                + " retries",
-            taskIdHex,
-            replicationType,
-            uri,
-            delay,
-            retryCount);
-      } else {
-        metrics.record(config.getName(), delay, retryCount);
-        long elapsed = NANOSECONDS.toMillis(context.stop());
-        Optional<Long> elapsedEnd2End =
-            apiRequestMetrics
-                .flatMap(metrics -> metrics.stop(config.getName()))
-                .map(NANOSECONDS::toMillis);
-        repLog.info(
-            "[{}] Replication from {} completed in {}ms, {}ms delay, {} retries{}",
-            taskIdHex,
-            uri,
-            elapsed,
-            delay,
-            retryCount,
-            elapsedEnd2End.map(el -> String.format(", E2E %dms", el)).orElse(""));
-      }
-    } catch (RepositoryNotFoundException e) {
-      stateLog.error(
-          "["
-              + taskIdHex
-              + "] Cannot replicate "
-              + projectName
-              + "; Local repository error: "
-              + e.getMessage(),
-          getStatesAsArray());
+        if (fetchRefSpecs.isEmpty()) {
+          repLog.info(
+              "[{}] {} replication from {} finished but no refs were replicated, {}ms delay, {}"
+                  + " retries",
+              taskIdHex,
+              replicationType,
+              uri,
+              delay,
+              retryCount);
+        } else {
+          metrics.record(config.getName(), delay, retryCount);
+          long elapsed = NANOSECONDS.toMillis(context.stop());
+          Optional<Long> elapsedEnd2End =
+              apiRequestMetrics
+                  .flatMap(metrics -> metrics.stop(config.getName()))
+                  .map(NANOSECONDS::toMillis);
+          repLog.info(
+              "[{}] Replication from {} completed in {}ms, {}ms delay, {} retries{}",
+              taskIdHex,
+              uri,
+              elapsed,
+              delay,
+              retryCount,
+              elapsedEnd2End.map(el -> String.format(", E2E %dms", el)).orElse(""));
+        }
+      } catch (RepositoryNotFoundException e) {
+        stateLog.error(
+            "["
+                + taskIdHex
+                + "] Cannot replicate "
+                + projectName
+                + "; Local repository error: "
+                + e.getMessage(),
+            getStatesAsArray());
 
-    } catch (NoRemoteRepositoryException | RemoteRepositoryException e) {
-      // Tried to replicate to a remote via anonymous git:// but the repository
-      // does not exist. In this case NoRemoteRepositoryException is not
-      // raised.
-      String msg = e.getMessage();
-      repLog.error(
-          "[{}] Cannot replicate {}; Remote repository error: {}", taskIdHex, projectName, msg);
-    } catch (NotSupportedException e) {
-      stateLog.error("[" + taskIdHex + "] Cannot replicate  from " + uri, e, getStatesAsArray());
-    } catch (PermanentTransportException e) {
-      repLog.error(
-          String.format("Terminal failure. Cannot replicate [%s] from %s", taskIdHex, uri), e);
-    } catch (TransportException e) {
-      repLog.error(String.format("[%s] Cannot replicate from %s", taskIdHex, uri), e);
-      if (replicationType == ReplicationType.ASYNC && e instanceof LockFailureException) {
-        lockRetryCount++;
-        // The LockFailureException message contains both URI and reason
-        // for this failure.
+      } catch (NoRemoteRepositoryException | RemoteRepositoryException e) {
+        // Tried to replicate to a remote via anonymous git:// but the repository
+        // does not exist. In this case NoRemoteRepositoryException is not
+        // raised.
+        String msg = e.getMessage();
+        repLog.error(
+            "[{}] Cannot replicate {}; Remote repository error: {}", taskIdHex, projectName, msg);
+      } catch (NotSupportedException e) {
+        stateLog.error("[" + taskIdHex + "] Cannot replicate  from " + uri, e, getStatesAsArray());
+      } catch (PermanentTransportException e) {
+        repLog.error(
+            String.format("Terminal failure. Cannot replicate [%s] from %s", taskIdHex, uri), e);
+      } catch (TransportException e) {
+        repLog.error(String.format("[%s] Cannot replicate from %s", taskIdHex, uri), e);
+        if (replicationType == ReplicationType.ASYNC && e instanceof LockFailureException) {
+          lockRetryCount++;
+          // The LockFailureException message contains both URI and reason
+          // for this failure.
 
-        // The remote fetch operation should be retried.
-        if (lockRetryCount <= maxLockRetries) {
+          // The remote fetch operation should be retried.
+          if (lockRetryCount <= maxLockRetries) {
+            if (canceledWhileRunning.get()) {
+              logCanceledWhileRunningException(e);
+            } else {
+              pool.reschedule(this, Source.RetryReason.TRANSPORT_ERROR);
+            }
+          } else {
+            repLog.error(
+                "[{}] Giving up after {} occurrences of this error: {} during replication from [{}]"
+                    + " {}",
+                taskIdHex,
+                lockRetryCount,
+                e.getMessage(),
+                taskIdHex,
+                uri);
+          }
+        } else if (replicationType == ReplicationType.ASYNC) {
           if (canceledWhileRunning.get()) {
             logCanceledWhileRunningException(e);
           } else {
+            // The remote fetch operation should be retried.
             pool.reschedule(this, Source.RetryReason.TRANSPORT_ERROR);
           }
-        } else {
-          repLog.error(
-              "[{}] Giving up after {} occurrences of this error: {} during replication from [{}]"
-                  + " {}",
-              taskIdHex,
-              lockRetryCount,
-              e.getMessage(),
-              taskIdHex,
-              uri);
         }
-      } else if (replicationType == ReplicationType.ASYNC) {
-        if (canceledWhileRunning.get()) {
-          logCanceledWhileRunningException(e);
-        } else {
-          // The remote fetch operation should be retried.
-          pool.reschedule(this, Source.RetryReason.TRANSPORT_ERROR);
+      } catch (IOException e) {
+        stateLog.error("[" + taskIdHex + "] Cannot replicate from " + uri, e, getStatesAsArray());
+      } catch (RuntimeException | Error e) {
+        stateLog.error(
+            "[" + taskIdHex + "] Unexpected error during replication from " + uri,
+            e,
+            getStatesAsArray());
+      } finally {
+        if (git != null) {
+          git.close();
         }
-      }
-    } catch (IOException e) {
-      stateLog.error("[" + taskIdHex + "] Cannot replicate from " + uri, e, getStatesAsArray());
-    } catch (RuntimeException | Error e) {
-      stateLog.error(
-          "[" + taskIdHex + "] Unexpected error during replication from " + uri,
-          e,
-          getStatesAsArray());
-    } finally {
-      if (git != null) {
-        git.close();
-      }
 
-      if (replicationType == ReplicationType.ASYNC) {
-        pool.notifyFinished(this);
+        if (replicationType == ReplicationType.ASYNC) {
+          pool.notifyFinished(this);
+        }
+      }
+    } catch (ProjectsLock.UnableToLockProjectException e) {
+      pool.fetchWasCanceled(this);
+      if (!canceled) {
+        repLog.info(
+            "[{}] Project {} is locked. Rescheduling replication from {} to avoid collision with an"
+                + " in-flight fetch task [{}].",
+            taskIdHex,
+            projectName,
+            uri,
+            e.getConflictingTaskId());
+        pool.reschedule(this, Source.RetryReason.COLLISION);
+        isCollision = true;
       }
     }
   }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/ProjectsLock.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/ProjectsLock.java
new file mode 100644
index 0000000..bfcc396
--- /dev/null
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/ProjectsLock.java
@@ -0,0 +1,80 @@
+// 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.replication.pull;
+
+import com.google.gerrit.entities.Project;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicReference;
+
+class ProjectsLock {
+  public static final String EMPTY_TASK_ID = "";
+  private final Map<Project.NameKey, AtomicReference<String>> projectLocksHeldByTasks =
+      new ConcurrentHashMap<>();
+
+  LockToken tryLock(Project.NameKey project, String taskId) throws UnableToLockProjectException {
+    if (!getProjectAtomicReference(project).compareAndSet(EMPTY_TASK_ID, taskId)) {
+      throw new UnableToLockProjectException(project, projectLocksHeldByTasks.get(project).get());
+    }
+    return new LockToken(this, project, taskId);
+  }
+
+  boolean unlock(Project.NameKey project, String taskId) {
+    return getProjectAtomicReference(project).compareAndSet(taskId, EMPTY_TASK_ID);
+  }
+
+  private AtomicReference<String> getProjectAtomicReference(Project.NameKey project) {
+    return projectLocksHeldByTasks.computeIfAbsent(
+        project, k -> new AtomicReference<>(EMPTY_TASK_ID));
+  }
+
+  static final class LockToken implements AutoCloseable {
+    final Project.NameKey project;
+    final ProjectsLock projectsLock;
+    private final String taskId;
+    boolean closed;
+
+    private LockToken(ProjectsLock projectsLock, Project.NameKey project, String taskId) {
+      this.project = project;
+      this.projectsLock = projectsLock;
+      this.taskId = taskId;
+    }
+
+    @Override
+    public void close() {
+      if (!closed && projectsLock.unlock(project, taskId)) {
+        closed = true;
+      }
+    }
+  }
+
+  static class UnableToLockProjectException extends Exception {
+    private final String conflictingTaskId;
+
+    public UnableToLockProjectException(Project.NameKey project, String conflictingTaskId) {
+      super(
+          "Unable to lock project "
+              + project
+              + " because it is already locked by task "
+              + conflictingTaskId);
+
+      this.conflictingTaskId = conflictingTaskId;
+    }
+
+    public String getConflictingTaskId() {
+      return conflictingTaskId;
+    }
+  }
+}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/PullReplicationModule.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/PullReplicationModule.java
index 251f39f..4f32f26 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/PullReplicationModule.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/PullReplicationModule.java
@@ -89,6 +89,7 @@
             .implement(HttpClient.class, SourceHttpClient.class)
             .build(SourceHttpClient.Factory.class));
 
+    bind(ProjectsLock.class).in(Scopes.SINGLETON);
     install(new FactoryModuleBuilder().build(Source.Factory.class));
     install(
         new FactoryModuleBuilder()
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/pull/FetchOneTest.java b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/FetchOneTest.java
index 1104ff4..50d91ff 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/replication/pull/FetchOneTest.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/FetchOneTest.java
@@ -130,6 +130,7 @@
             fetchFactory,
             fetchRefsDatabase,
             deleteRefCommand,
+            new ProjectsLock(),
             PROJECT_NAME,
             urIish,
             Optional.of(pullReplicationApiRequestMetrics));
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/pull/ProjectsLockTest.java b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/ProjectsLockTest.java
new file mode 100644
index 0000000..89d1295
--- /dev/null
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/ProjectsLockTest.java
@@ -0,0 +1,65 @@
+// 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.replication.pull;
+
+import static com.google.common.truth.Truth.assertThat;
+import static com.google.gerrit.testing.GerritJUnit.assertThrows;
+
+import com.google.gerrit.entities.Project;
+import org.junit.Test;
+
+public class ProjectsLockTest {
+  private final ProjectsLock projectsLock = new ProjectsLock();
+  private final Project.NameKey project1 = Project.nameKey("project1");
+  private final Project.NameKey project2 = Project.nameKey("project2");
+  private final String task1 = "task1";
+  private final String task2 = "task2";
+
+  @Test
+  public void shouldSuccessfullyLockAndUnlockProject() throws Exception {
+    ProjectsLock.LockToken lockToken = projectsLock.tryLock(project1, task1);
+    assertThatIsLocked(lockToken, project1);
+    assertThat(projectsLock.unlock(project1, task1)).isTrue();
+
+    try (ProjectsLock.LockToken newLockToken = projectsLock.tryLock(project1, task2)) {
+      assertThatIsLocked(newLockToken, project1);
+    }
+  }
+
+  @Test
+  public void shouldFailToLockFromAnotherTaskOnLockedProject() throws Exception {
+    try (ProjectsLock.LockToken unused = projectsLock.tryLock(project1, task1)) {
+      ProjectsLock.UnableToLockProjectException e =
+          assertThrows(
+              ProjectsLock.UnableToLockProjectException.class,
+              () -> projectsLock.tryLock(project1, task2).close());
+      assertThat(e.getConflictingTaskId()).isEqualTo(task1);
+    }
+  }
+
+  @Test
+  public void shouldHandleMultipleProjectsIndependently() throws Exception {
+    try (ProjectsLock.LockToken lockToken1 = projectsLock.tryLock(project1, task1);
+        ProjectsLock.LockToken lockToken2 = projectsLock.tryLock(project2, task2)) {
+      assertThatIsLocked(lockToken1, project1);
+      assertThatIsLocked(lockToken2, project2);
+    }
+  }
+
+  private void assertThatIsLocked(ProjectsLock.LockToken lockToken, Project.NameKey project1) {
+    assertThat(lockToken).isNotNull();
+    assertThat(lockToken.project).isEqualTo(project1);
+  }
+}