diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/Destination.java b/src/main/java/com/googlesource/gerrit/plugins/replication/Destination.java
index dbdb325..4ebd67e 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/Destination.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/Destination.java
@@ -60,7 +60,6 @@
 import com.google.inject.assistedinject.FactoryModuleBuilder;
 import com.google.inject.servlet.RequestScoped;
 import com.googlesource.gerrit.plugins.replication.ReplicationState.RefPushResult;
-import com.googlesource.gerrit.plugins.replication.ReplicationTasksStorage.ReplicateRefUpdate;
 import java.io.IOException;
 import java.io.UnsupportedEncodingException;
 import java.net.URLEncoder;
@@ -522,6 +521,7 @@
         pending.put(uri, pushOp);
         switch (reason) {
           case COLLISION:
+            replicationTasksStorage.get().reset(pushOp);
             @SuppressWarnings("unused")
             ScheduledFuture<?> ignored =
                 pool.schedule(pushOp, config.getRescheduleDelay(), TimeUnit.SECONDS);
@@ -536,6 +536,7 @@
             postReplicationFailedEvent(pushOp, status);
             if (pushOp.setToRetry()) {
               postReplicationScheduledEvent(pushOp);
+              replicationTasksStorage.get().reset(pushOp);
               @SuppressWarnings("unused")
               ScheduledFuture<?> ignored2 =
                   pool.schedule(pushOp, config.getRetryDelay(), TimeUnit.MINUTES);
@@ -562,25 +563,16 @@
       if (inFlightOp != null) {
         return RunwayStatus.denied(inFlightOp.getId());
       }
+      replicationTasksStorage.get().start(op);
       inFlight.put(op.getURI(), op);
     }
     return RunwayStatus.allowed();
   }
 
-  void notifyFinished(PushOne task) {
+  void notifyFinished(PushOne op) {
     synchronized (stateLock) {
-      inFlight.remove(task.getURI());
-      if (!task.wasCanceled()) {
-        for (String ref : task.getRefs()) {
-          if (!refHasPendingPush(task.getURI(), ref)) {
-            replicationTasksStorage
-                .get()
-                .delete(
-                    new ReplicateRefUpdate(
-                        task.getProjectNameKey().get(), ref, task.getURI(), getRemoteConfigName()));
-          }
-        }
-      }
+      replicationTasksStorage.get().finish(op);
+      inFlight.remove(op.getURI());
     }
   }
 
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 e02fafc..5dc8f73 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationQueue.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationQueue.java
@@ -78,6 +78,7 @@
     if (!running) {
       destinations.get().startup(workQueue);
       running = true;
+      replicationTasksStorage.resetAll();
       firePendingEvents();
       fireBeforeStartupEvents();
     }
@@ -153,7 +154,7 @@
     try {
       Set<String> eventsReplayed = new HashSet<>();
       replaying = true;
-      for (ReplicationTasksStorage.ReplicateRefUpdate t : replicationTasksStorage.list()) {
+      for (ReplicationTasksStorage.ReplicateRefUpdate t : replicationTasksStorage.listWaiting()) {
         String eventKey = String.format("%s:%s", t.project, t.ref);
         if (!eventsReplayed.contains(eventKey)) {
           repLog.info("Firing pending task {}", eventKey);
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationTasksStorage.java b/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationTasksStorage.java
index 3b174a3..950ef26 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationTasksStorage.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationTasksStorage.java
@@ -27,11 +27,32 @@
 import java.nio.file.DirectoryStream;
 import java.nio.file.Files;
 import java.nio.file.Path;
+import java.nio.file.StandardCopyOption;
 import java.util.ArrayList;
 import java.util.List;
 import org.eclipse.jgit.lib.ObjectId;
 import org.eclipse.jgit.transport.URIish;
 
+/**
+ * A persistent store for replication tasks.
+ *
+ * <p>The data of this store lives under <replication_data>/ref-updates where replication_data is
+ * determined by the replication.eventsDirectory config option and defaults to
+ * <site_dir>/data/replication. Atomic renames must be supported from anywhere within the store to
+ * anywhere within the store. This generally means that all the contents of the store needs to live
+ * on the same filesystem.
+ *
+ * <p>Individual tasks are stored in files under the following directories using the sha1 of the
+ * task:
+ *
+ * <p><code>
+ *   .../running/<sha1>    running replication tasks
+ *   .../waiting/<sha1>    outstanding replication tasks
+ * </code>
+ *
+ * <p>Tasks are moved atomically via a rename between those directories to indicate the current
+ * state of each task.
+ */
 @Singleton
 public class ReplicationTasksStorage {
   private static final FluentLogger logger = FluentLogger.forEnclosingClass();
@@ -44,6 +65,10 @@
     public final String uri;
     public final String remote;
 
+    public ReplicateRefUpdate(PushOne push, String ref) {
+      this(push.getProjectNameKey().get(), ref, push.getURI(), push.getRemoteName());
+    }
+
     public ReplicateRefUpdate(String project, String ref, URIish uri, String remote) {
       this.project = project;
       this.ref = ref;
@@ -59,11 +84,14 @@
 
   private static Gson GSON = new Gson();
 
-  private final Path refUpdates;
+  private final Path runningUpdates;
+  private final Path waitingUpdates;
 
   @Inject
   ReplicationTasksStorage(ReplicationConfig config) {
-    refUpdates = config.getEventsDirectory().resolve("ref-updates");
+    Path refUpdates = config.getEventsDirectory().resolve("ref-updates");
+    runningUpdates = refUpdates.resolve("running");
+    waitingUpdates = refUpdates.resolve("waiting");
   }
 
   public String persist(ReplicateRefUpdate r) {
@@ -75,13 +103,37 @@
     this.disableDeleteForTesting = deleteDisabled;
   }
 
-  public void delete(ReplicateRefUpdate r) {
-    new Task(r).delete();
+  public void start(PushOne push) {
+    for (String ref : push.getRefs()) {
+      new Task(new ReplicateRefUpdate(push, ref)).start();
+    }
   }
 
-  public List<ReplicateRefUpdate> list() {
+  public void reset(PushOne push) {
+    for (String ref : push.getRefs()) {
+      new Task(new ReplicateRefUpdate(push, ref)).reset();
+    }
+  }
+
+  public void resetAll() {
+    for (ReplicateRefUpdate r : list(createDir(runningUpdates))) {
+      new Task(r).reset();
+    }
+  }
+
+  public void finish(PushOne push) {
+    for (String ref : push.getRefs()) {
+      new Task(new ReplicateRefUpdate(push, ref)).finish();
+    }
+  }
+
+  public List<ReplicateRefUpdate> listWaiting() {
+    return list(createDir(waitingUpdates));
+  }
+
+  private List<ReplicateRefUpdate> list(Path tasks) {
     List<ReplicateRefUpdate> results = new ArrayList<>();
-    try (DirectoryStream<Path> events = Files.newDirectoryStream(refUpdates())) {
+    try (DirectoryStream<Path> events = Files.newDirectoryStream(tasks)) {
       for (Path e : events) {
         if (Files.isRegularFile(e)) {
           String json = new String(Files.readAllBytes(e), UTF_8);
@@ -89,7 +141,7 @@
         }
       }
     } catch (IOException e) {
-      logger.atSevere().withCause(e).log("Error when firing pending tasks");
+      logger.atSevere().withCause(e).log("Error while listing tasks");
     }
     return results;
   }
@@ -99,11 +151,11 @@
     return ObjectId.fromRaw(Hashing.sha1().hashString(s, UTF_8).asBytes());
   }
 
-  private Path refUpdates() {
+  private static Path createDir(Path dir) {
     try {
-      return Files.createDirectories(refUpdates);
+      return Files.createDirectories(dir);
     } catch (IOException e) {
-      throw new ProvisionException(String.format("Couldn't create %s", refUpdates), e);
+      throw new ProvisionException(String.format("Couldn't create %s", dir), e);
     }
   }
 
@@ -111,44 +163,63 @@
     public final ReplicateRefUpdate update;
     public final String json;
     public final String taskKey;
-    public final Path file;
+    public final Path running;
+    public final Path waiting;
 
     public Task(ReplicateRefUpdate update) {
       this.update = update;
       json = GSON.toJson(update) + "\n";
       String key = update.project + "\n" + update.ref + "\n" + update.uri + "\n" + update.remote;
       taskKey = sha1(key).name();
-      file = refUpdates().resolve(taskKey);
+      running = createDir(runningUpdates).resolve(taskKey);
+      waiting = createDir(waitingUpdates).resolve(taskKey);
     }
 
     public String persist() {
-      if (Files.exists(file)) {
+      if (Files.exists(waiting)) {
         return taskKey;
       }
 
       try {
-        logger.atFine().log("CREATE %s %s", file, updateLog());
-        Files.write(file, json.getBytes(UTF_8));
+        logger.atFine().log("CREATE %s %s", waiting, updateLog());
+        Files.write(waiting, json.getBytes(UTF_8));
       } catch (IOException e) {
         logger.atWarning().withCause(e).log("Couldn't persist task %s", json);
       }
       return taskKey;
     }
 
-    public void delete() {
+    public void start() {
+      rename(waiting, running);
+    }
+
+    public void reset() {
+      rename(running, waiting);
+    }
+
+    public void finish() {
       if (disableDeleteForTesting) {
-        logger.atFine().log("DELETE %s %s DISABLED", file, updateLog());
+        logger.atFine().log("DELETE %s %s DISABLED", running, updateLog());
         return;
       }
 
       try {
-        logger.atFine().log("DELETE %s %s", file, updateLog());
-        Files.delete(file);
+        logger.atFine().log("DELETE %s %s", running, updateLog());
+        Files.delete(running);
       } catch (IOException e) {
         logger.atSevere().withCause(e).log("Error while deleting task %s", taskKey);
       }
     }
 
+    private void rename(Path from, Path to) {
+      try {
+        logger.atFine().log("RENAME %s to %s %s", from, to, updateLog());
+        Files.move(from, to, StandardCopyOption.ATOMIC_MOVE, StandardCopyOption.REPLACE_EXISTING);
+      } catch (IOException e) {
+        logger.atSevere().withCause(e).log("Error while renaming task %s", taskKey);
+      }
+    }
+
     private String updateLog() {
       return String.format("(%s:%s => %s)", update.project, update.ref, update.uri);
     }
diff --git a/src/main/resources/Documentation/config.md b/src/main/resources/Documentation/config.md
index d9d4bae..7233061 100644
--- a/src/main/resources/Documentation/config.md
+++ b/src/main/resources/Documentation/config.md
@@ -142,6 +142,11 @@
 	persisted events will not be deleted. When the plugin is started again,
 	it will trigger all replications found under this directory.
 
+	For replication to work, is is important that atomic renames be possible
+	from within any subdirectory of the eventsDirectory to within any other
+	subdirectory of the eventsDirectory. This generally means that the entire
+	contents of the eventsDirectory should live on the same filesystem.
+
 	When not set, defaults to the plugin's data directory.
 
 remote.NAME.url
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationIT.java b/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationIT.java
index 3b7a3ed..19269f0 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationIT.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationIT.java
@@ -399,7 +399,7 @@
 
   private List<ReplicateRefUpdate> listReplicationTasks(String refRegex) {
     Pattern refmaskPattern = Pattern.compile(refRegex);
-    return tasksStorage.list().stream()
+    return tasksStorage.listWaiting().stream()
         .filter(task -> refmaskPattern.matcher(task.ref).matches())
         .collect(toList());
   }