src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationTasksStorage.java - plugins/replication - Git at Google

 // Copyright (C) 2018 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;

 import static java.nio.charset.StandardCharsets.UTF_8;

 import com.google.auto.value.AutoValue;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.ImmutableSet;
 import com.google.common.flogger.FluentLogger;
 import com.google.common.hash.Hashing;
 import com.google.gerrit.common.Nullable;
 import com.google.gson.Gson;
 import com.google.gson.GsonBuilder;
 import com.google.gson.TypeAdapter;
 import com.google.gson.TypeAdapterFactory;
 import com.google.gson.reflect.TypeToken;
 import com.google.gson.stream.JsonReader;
 import com.google.gson.stream.JsonToken;
 import com.google.gson.stream.JsonWriter;
 import com.google.inject.Inject;
 import com.google.inject.ProvisionException;
 import com.google.inject.Singleton;
 import com.googlesource.gerrit.plugins.replication.api.ReplicationConfig;
 import java.io.IOException;
 import java.net.URISyntaxException;
 import java.nio.file.Files;
 import java.nio.file.NoSuchFileException;
 import java.nio.file.NotDirectoryException;
 import java.nio.file.Path;
 import java.nio.file.StandardCopyOption;
 import java.util.HashSet;
 import java.util.Optional;
 import java.util.Set;
 import java.util.stream.Stream;
 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>
  *   .../building/<tmp_name>                       new replication tasks under construction
  *   .../running/<sha1>                            running replication tasks
  *   .../waiting/<task_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();

   @AutoValue
   public abstract static class ReplicateRefUpdate {
     public static Optional<ReplicateRefUpdate> createOptionally(Path file, Gson gson) {
       try {
         return Optional.ofNullable(create(file, gson));
       } catch (NoSuchFileException e) {
         logger.atFine().log("File %s not found while reading task", file);
       } catch (IOException e) {
         logger.atSevere().withCause(e).log("Error while reading task %s", file);
       }
       return Optional.empty();
     }

     public static ReplicateRefUpdate create(Path file, Gson gson) throws IOException {
       String json = new String(Files.readAllBytes(file), UTF_8);
       return create(gson.fromJson(json, ReplicateRefUpdate.class), file.getFileName().toString());
     }

     public static ReplicateRefUpdate create(
         String project, Set<String> refs, URIish uri, String remote) {
       return new AutoValue_ReplicationTasksStorage_ReplicateRefUpdate(
           project,
           ImmutableSet.copyOf(refs),
           uri.toASCIIString(),
           remote,
           sha1(project, ImmutableSet.copyOf(refs), uri.toASCIIString(), remote));
     }

     public static ReplicateRefUpdate create(ReplicateRefUpdate u, String filename) {
       return new AutoValue_ReplicationTasksStorage_ReplicateRefUpdate(
           u.project(), u.refs(), u.uri(), u.remote(), filename);
     }

     public abstract String project();

     public abstract ImmutableSet<String> refs();

     public abstract String uri();

     public abstract String remote();

     public abstract String sha1();

     private static String sha1(String project, Set<String> refs, String uri, String remote) {
       return ReplicationTasksStorage.sha1(
               project + "\n" + refs.toString() + "\n" + uri + "\n" + remote)
           .name();
     }

     @Override
     public final String toString() {
       return "ref-update "
           + project()
           + ":"
           + refs().toString()
           + " uri:"
           + uri()
           + " remote:"
           + remote();
     }

     public static TypeAdapter<ReplicateRefUpdate> typeAdapter(Gson gson) {
       return new AutoValue_ReplicationTasksStorage_ReplicateRefUpdate.GsonTypeAdapter(gson);
     }
   }

   private final Gson gson;

   private final Path buildingUpdates;
   private final Path runningUpdates;
   private final Path waitingUpdates;

   private boolean isMultiPrimary;

   @Inject
   ReplicationTasksStorage(ReplicationConfig config) {
     this(config.getEventsDirectory().resolve("ref-updates"));
     isMultiPrimary = config.getDistributionInterval() != 0;
   }

   @VisibleForTesting
   public ReplicationTasksStorage(Path refUpdates) {
     buildingUpdates = refUpdates.resolve("building");
     runningUpdates = refUpdates.resolve("running");
     waitingUpdates = refUpdates.resolve("waiting");
     gson =
         new GsonBuilder()
             .registerTypeAdapterFactory(new ReplicateRefUpdateTypeAdapterFactory())
             .create();
   }

   private boolean isMultiPrimary() {
     return isMultiPrimary;
   }

   public synchronized String create(ReplicateRefUpdate r) {
     return new Task(r).create();
   }

   public synchronized Set<ImmutableSet<String>> start(UriUpdates uriUpdates) {
     Set<ImmutableSet<String>> startedRefs = new HashSet<>();
     for (ReplicateRefUpdate update : uriUpdates.getReplicateRefUpdates()) {
       Task t = new Task(update);
       if (t.start()) {
         startedRefs.add(t.update.refs());
       }
     }
     return startedRefs;
   }

   public synchronized void reset(UriUpdates uriUpdates) {
     for (ReplicateRefUpdate update : uriUpdates.getReplicateRefUpdates()) {
       new Task(update).reset();
     }
   }

   public synchronized void recoverAll() {
     streamRunning().forEach(r -> new Task(r).recover());
   }

   public boolean isWaiting(UriUpdates uriUpdates) {
     return uriUpdates.getReplicateRefUpdates().stream()
         .map(update -> new Task(update))
         .anyMatch(Task::isWaiting);
   }

   public void finish(UriUpdates uriUpdates) {
     for (ReplicateRefUpdate update : uriUpdates.getReplicateRefUpdates()) {
       new Task(update).finish();
     }
   }

   public Stream<ReplicateRefUpdate> streamWaiting() {
     return streamRecursive(createDir(waitingUpdates));
   }

   public Stream<ReplicateRefUpdate> streamRunning() {
     return streamRecursive(createDir(runningUpdates));
   }

   private Stream<ReplicateRefUpdate> streamRecursive(Path dir) {
     return walkNonDirs(dir)
         .map(path -> ReplicateRefUpdate.createOptionally(path, gson))
         .filter(Optional::isPresent)
         .map(Optional::get);
   }

   private Stream<Path> walkNonDirs(Path path) {
     try {
       return Files.list(path).flatMap(sub -> walkNonDirs(sub));
     } catch (NotDirectoryException e) {
       return Stream.of(path);
     } catch (Exception e) {
       String message = "Error while walking directory %s";
       if (isMultiPrimary() && e instanceof NoSuchFileException) {
         logger.atFine().log(
             message + " (expected regularly with multi-primaries and distributor enabled)", path);
       } else {
         logger.atSevere().withCause(e).log(message, path);
       }
       return Stream.empty();
     }
   }

   @SuppressWarnings("deprecation")
   private static ObjectId sha1(String s) {
     return ObjectId.fromRaw(Hashing.sha1().hashString(s, UTF_8).asBytes());
   }

   private static Path createDir(Path dir) {
     try {
       return Files.createDirectories(dir);
     } catch (IOException e) {
       throw new ProvisionException(String.format("Couldn't create %s", dir), e);
     }
   }

   public static final class ReplicateRefUpdateTypeAdapterFactory implements TypeAdapterFactory {
     static class ReplicateRefUpdateTypeAdapter<T> extends TypeAdapter<ReplicateRefUpdate> {

       @Override
       public void write(JsonWriter out, ReplicateRefUpdate value) throws IOException {
         if (value == null) {
           out.nullValue();
           return;
         }
         out.beginObject();

         out.name("project");
         out.value(value.project());

         out.name("refs");
         out.beginArray();
         for (String ref : value.refs()) {
           out.value(ref);
         }
         out.endArray();

         out.name("uri");
         out.value(value.uri());

         out.name("remote");
         out.value(value.remote());

         out.endObject();
       }

       @Nullable
       @Override
       public ReplicateRefUpdate read(JsonReader in) throws IOException {
         if (in.peek() == JsonToken.NULL) {
           in.nextNull();
           return null;
         }
         String project = null;
         Set<String> refs = new HashSet<>();
         URIish uri = null;
         String remote = null;

         String fieldname = null;
         in.beginObject();

         while (in.hasNext()) {
           JsonToken token = in.peek();

           if (token.equals(JsonToken.NAME)) {
             fieldname = in.nextName();
           }

           switch (fieldname) {
             case "project":
               project = in.nextString();
               break;
             case "refs":
               in.beginArray();
               while (in.hasNext()) {
                 refs.add(in.nextString());
               }
               in.endArray();
               break;
             case "ref":
               refs.add(in.nextString());
               break;
             case "uri":
               try {
                 uri = new URIish(in.nextString());
               } catch (URISyntaxException e) {
                 throw new IOException("Unable to parse remote URI", e);
               }
               break;
             case "remote":
               remote = in.nextString();
               break;
             default:
               throw new IOException(String.format("Unknown field in stored task: %s", fieldname));
           }
         }

         in.endObject();
         return ReplicateRefUpdate.create(project, refs, uri, remote);
       }
     }

     @SuppressWarnings("unchecked")
     @Nullable
     @Override
     public <T> TypeAdapter<T> create(Gson gson, TypeToken<T> type) {
       if (type.equals(TypeToken.get(AutoValue_ReplicationTasksStorage_ReplicateRefUpdate.class))
           || type.equals(TypeToken.get(ReplicateRefUpdate.class))) {
         return (TypeAdapter<T>) new ReplicateRefUpdateTypeAdapter<T>();
       }
       return null;
     }
   }

   @VisibleForTesting
   class Task {
     public final ReplicateRefUpdate update;
     public final String taskKey;
     public final Path running;
     public final Path waiting;

     public Task(ReplicateRefUpdate update) {
       this.update = update;
       taskKey = update.sha1();
       running = createDir(runningUpdates).resolve(taskKey);
       waiting = createDir(waitingUpdates).resolve(taskKey);
     }

     public String create() {
       if (Files.exists(waiting)) {
         return taskKey;
       }

       String json = gson.toJson(update) + "\n";
       try {
         Path tmp = Files.createTempFile(createDir(buildingUpdates), taskKey, null);
         logger.atFine().log("CREATE %s %s", tmp, updateLog());
         Files.write(tmp, json.getBytes(UTF_8));
         logger.atFine().log("RENAME %s %s %s", tmp, waiting, updateLog());
         rename(tmp, waiting);
       } catch (IOException e) {
         logger.atWarning().withCause(e).log("Couldn't create task %s", json);
       }
       return taskKey;
     }

     public boolean start() {
       return rename(waiting, running);
     }

     public void reset() {
       rename(running, waiting);
     }

     public void recover() {
       rename(running, waiting);
     }

     public boolean isWaiting() {
       return Files.exists(waiting);
     }

     public void finish() {
       try {
         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);
       }
     }

     @VisibleForTesting
     boolean 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);
         return true;
       } catch (IOException e) {
         String message = "Error while renaming task %s";
         if (isMultiPrimary() && e instanceof NoSuchFileException) {
           logger.atFine().log(
               message + " (expected regularly with multi-primaries and distributor enabled)",
               taskKey);
         } else {
           logger.atSevere().withCause(e).log(message, taskKey);
         }
         return false;
       }
     }

     private String updateLog() {
       return String.format("(%s:%s => %s)", update.project(), update.refs(), update.uri());
     }
   }
 }
	// Copyright (C) 2018 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;

	import static java.nio.charset.StandardCharsets.UTF_8;

	import com.google.auto.value.AutoValue;
	import com.google.common.annotations.VisibleForTesting;
	import com.google.common.collect.ImmutableSet;
	import com.google.common.flogger.FluentLogger;
	import com.google.common.hash.Hashing;
	import com.google.gerrit.common.Nullable;
	import com.google.gson.Gson;
	import com.google.gson.GsonBuilder;
	import com.google.gson.TypeAdapter;
	import com.google.gson.TypeAdapterFactory;
	import com.google.gson.reflect.TypeToken;
	import com.google.gson.stream.JsonReader;
	import com.google.gson.stream.JsonToken;
	import com.google.gson.stream.JsonWriter;
	import com.google.inject.Inject;
	import com.google.inject.ProvisionException;
	import com.google.inject.Singleton;
	import com.googlesource.gerrit.plugins.replication.api.ReplicationConfig;
	import java.io.IOException;
	import java.net.URISyntaxException;
	import java.nio.file.Files;
	import java.nio.file.NoSuchFileException;
	import java.nio.file.NotDirectoryException;
	import java.nio.file.Path;
	import java.nio.file.StandardCopyOption;
	import java.util.HashSet;
	import java.util.Optional;
	import java.util.Set;
	import java.util.stream.Stream;
	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>
	* .../building/<tmp_name> new replication tasks under construction
	* .../running/<sha1> running replication tasks
	* .../waiting/<task_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();

	@AutoValue
	public abstract static class ReplicateRefUpdate {
	public static Optional<ReplicateRefUpdate> createOptionally(Path file, Gson gson) {
	try {
	return Optional.ofNullable(create(file, gson));
	} catch (NoSuchFileException e) {
	logger.atFine().log("File %s not found while reading task", file);
	} catch (IOException e) {
	logger.atSevere().withCause(e).log("Error while reading task %s", file);
	}
	return Optional.empty();
	}

	public static ReplicateRefUpdate create(Path file, Gson gson) throws IOException {
	String json = new String(Files.readAllBytes(file), UTF_8);
	return create(gson.fromJson(json, ReplicateRefUpdate.class), file.getFileName().toString());
	}

	public static ReplicateRefUpdate create(
	String project, Set<String> refs, URIish uri, String remote) {
	return new AutoValue_ReplicationTasksStorage_ReplicateRefUpdate(
	project,
	ImmutableSet.copyOf(refs),
	uri.toASCIIString(),
	remote,
	sha1(project, ImmutableSet.copyOf(refs), uri.toASCIIString(), remote));
	}

	public static ReplicateRefUpdate create(ReplicateRefUpdate u, String filename) {
	return new AutoValue_ReplicationTasksStorage_ReplicateRefUpdate(
	u.project(), u.refs(), u.uri(), u.remote(), filename);
	}

	public abstract String project();

	public abstract ImmutableSet<String> refs();

	public abstract String uri();

	public abstract String remote();

	public abstract String sha1();

	private static String sha1(String project, Set<String> refs, String uri, String remote) {
	return ReplicationTasksStorage.sha1(
	project + "\n" + refs.toString() + "\n" + uri + "\n" + remote)
	.name();
	}

	@Override
	public final String toString() {
	return "ref-update "
	+ project()
	+ ":"
	+ refs().toString()
	+ " uri:"
	+ uri()
	+ " remote:"
	+ remote();
	}

	public static TypeAdapter<ReplicateRefUpdate> typeAdapter(Gson gson) {
	return new AutoValue_ReplicationTasksStorage_ReplicateRefUpdate.GsonTypeAdapter(gson);
	}
	}

	private final Gson gson;

	private final Path buildingUpdates;
	private final Path runningUpdates;
	private final Path waitingUpdates;

	private boolean isMultiPrimary;

	@Inject
	ReplicationTasksStorage(ReplicationConfig config) {
	this(config.getEventsDirectory().resolve("ref-updates"));
	isMultiPrimary = config.getDistributionInterval() != 0;
	}

	@VisibleForTesting
	public ReplicationTasksStorage(Path refUpdates) {
	buildingUpdates = refUpdates.resolve("building");
	runningUpdates = refUpdates.resolve("running");
	waitingUpdates = refUpdates.resolve("waiting");
	gson =
	new GsonBuilder()
	.registerTypeAdapterFactory(new ReplicateRefUpdateTypeAdapterFactory())
	.create();
	}

	private boolean isMultiPrimary() {
	return isMultiPrimary;
	}

	public synchronized String create(ReplicateRefUpdate r) {
	return new Task(r).create();
	}

	public synchronized Set<ImmutableSet<String>> start(UriUpdates uriUpdates) {
	Set<ImmutableSet<String>> startedRefs = new HashSet<>();
	for (ReplicateRefUpdate update : uriUpdates.getReplicateRefUpdates()) {
	Task t = new Task(update);
	if (t.start()) {
	startedRefs.add(t.update.refs());
	}
	}
	return startedRefs;
	}

	public synchronized void reset(UriUpdates uriUpdates) {
	for (ReplicateRefUpdate update : uriUpdates.getReplicateRefUpdates()) {
	new Task(update).reset();
	}
	}

	public synchronized void recoverAll() {
	streamRunning().forEach(r -> new Task(r).recover());
	}

	public boolean isWaiting(UriUpdates uriUpdates) {
	return uriUpdates.getReplicateRefUpdates().stream()
	.map(update -> new Task(update))
	.anyMatch(Task::isWaiting);
	}

	public void finish(UriUpdates uriUpdates) {
	for (ReplicateRefUpdate update : uriUpdates.getReplicateRefUpdates()) {
	new Task(update).finish();
	}
	}

	public Stream<ReplicateRefUpdate> streamWaiting() {
	return streamRecursive(createDir(waitingUpdates));
	}

	public Stream<ReplicateRefUpdate> streamRunning() {
	return streamRecursive(createDir(runningUpdates));
	}

	private Stream<ReplicateRefUpdate> streamRecursive(Path dir) {
	return walkNonDirs(dir)
	.map(path -> ReplicateRefUpdate.createOptionally(path, gson))
	.filter(Optional::isPresent)
	.map(Optional::get);
	}

	private Stream<Path> walkNonDirs(Path path) {
	try {
	return Files.list(path).flatMap(sub -> walkNonDirs(sub));
	} catch (NotDirectoryException e) {
	return Stream.of(path);
	} catch (Exception e) {
	String message = "Error while walking directory %s";
	if (isMultiPrimary() && e instanceof NoSuchFileException) {
	logger.atFine().log(
	message + " (expected regularly with multi-primaries and distributor enabled)", path);
	} else {
	logger.atSevere().withCause(e).log(message, path);
	}
	return Stream.empty();
	}
	}

	@SuppressWarnings("deprecation")
	private static ObjectId sha1(String s) {
	return ObjectId.fromRaw(Hashing.sha1().hashString(s, UTF_8).asBytes());
	}

	private static Path createDir(Path dir) {
	try {
	return Files.createDirectories(dir);
	} catch (IOException e) {
	throw new ProvisionException(String.format("Couldn't create %s", dir), e);
	}
	}

	public static final class ReplicateRefUpdateTypeAdapterFactory implements TypeAdapterFactory {
	static class ReplicateRefUpdateTypeAdapter<T> extends TypeAdapter<ReplicateRefUpdate> {

	@Override
	public void write(JsonWriter out, ReplicateRefUpdate value) throws IOException {
	if (value == null) {
	out.nullValue();
	return;
	}
	out.beginObject();

	out.name("project");
	out.value(value.project());

	out.name("refs");
	out.beginArray();
	for (String ref : value.refs()) {
	out.value(ref);
	}
	out.endArray();

	out.name("uri");
	out.value(value.uri());

	out.name("remote");
	out.value(value.remote());

	out.endObject();
	}

	@Nullable
	@Override
	public ReplicateRefUpdate read(JsonReader in) throws IOException {
	if (in.peek() == JsonToken.NULL) {
	in.nextNull();
	return null;
	}
	String project = null;
	Set<String> refs = new HashSet<>();
	URIish uri = null;
	String remote = null;

	String fieldname = null;
	in.beginObject();

	while (in.hasNext()) {
	JsonToken token = in.peek();

	if (token.equals(JsonToken.NAME)) {
	fieldname = in.nextName();
	}

	switch (fieldname) {
	case "project":
	project = in.nextString();
	break;
	case "refs":
	in.beginArray();
	while (in.hasNext()) {
	refs.add(in.nextString());
	}
	in.endArray();
	break;
	case "ref":
	refs.add(in.nextString());
	break;
	case "uri":
	try {
	uri = new URIish(in.nextString());
	} catch (URISyntaxException e) {
	throw new IOException("Unable to parse remote URI", e);
	}
	break;
	case "remote":
	remote = in.nextString();
	break;
	default:
	throw new IOException(String.format("Unknown field in stored task: %s", fieldname));
	}
	}

	in.endObject();
	return ReplicateRefUpdate.create(project, refs, uri, remote);
	}
	}

	@SuppressWarnings("unchecked")
	@Nullable
	@Override
	public <T> TypeAdapter<T> create(Gson gson, TypeToken<T> type) {
	if (type.equals(TypeToken.get(AutoValue_ReplicationTasksStorage_ReplicateRefUpdate.class))
	\|\| type.equals(TypeToken.get(ReplicateRefUpdate.class))) {
	return (TypeAdapter<T>) new ReplicateRefUpdateTypeAdapter<T>();
	}
	return null;
	}
	}

	@VisibleForTesting
	class Task {
	public final ReplicateRefUpdate update;
	public final String taskKey;
	public final Path running;
	public final Path waiting;

	public Task(ReplicateRefUpdate update) {
	this.update = update;
	taskKey = update.sha1();
	running = createDir(runningUpdates).resolve(taskKey);
	waiting = createDir(waitingUpdates).resolve(taskKey);
	}

	public String create() {
	if (Files.exists(waiting)) {
	return taskKey;
	}

	String json = gson.toJson(update) + "\n";
	try {
	Path tmp = Files.createTempFile(createDir(buildingUpdates), taskKey, null);
	logger.atFine().log("CREATE %s %s", tmp, updateLog());
	Files.write(tmp, json.getBytes(UTF_8));
	logger.atFine().log("RENAME %s %s %s", tmp, waiting, updateLog());
	rename(tmp, waiting);
	} catch (IOException e) {
	logger.atWarning().withCause(e).log("Couldn't create task %s", json);
	}
	return taskKey;
	}

	public boolean start() {
	return rename(waiting, running);
	}

	public void reset() {
	rename(running, waiting);
	}

	public void recover() {
	rename(running, waiting);
	}

	public boolean isWaiting() {
	return Files.exists(waiting);
	}

	public void finish() {
	try {
	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);
	}
	}

	@VisibleForTesting
	boolean 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);
	return true;
	} catch (IOException e) {
	String message = "Error while renaming task %s";
	if (isMultiPrimary() && e instanceof NoSuchFileException) {
	logger.atFine().log(
	message + " (expected regularly with multi-primaries and distributor enabled)",
	taskKey);
	} else {
	logger.atSevere().withCause(e).log(message, taskKey);
	}
	return false;
	}
	}

	private String updateLog() {
	return String.format("(%s:%s => %s)", update.project(), update.refs(), update.uri());
	}
	}
	}