Allow RdeStagingAction to be invoked manually

RdeStagingAction always processed all RDE and BRDA deposits currently outstanding, updating the cursors appropriately and kicking off the upload job. Sometimes we don't want all that. We just want to create a specific deposit by hand, without modifying the cursors or uploading. This CL adds parameters to support that.

-------------
Created by MOE: https://github.com/google/moe
MOE_MIGRATED_REVID=152415959

java/google/registry/module/backend/BackendModule.java

@@ -15,10 +15,13 @@
 package google.registry.module.backend;
 
 import static google.registry.model.registry.Registries.assertTldExists;
+import static google.registry.model.registry.Registries.assertTldsExist;
 import static google.registry.request.RequestParameters.extractOptionalDatetimeParameter;
 import static google.registry.request.RequestParameters.extractRequiredParameter;
+import static google.registry.request.RequestParameters.extractSetOfParameters;
 
 import com.google.common.base.Optional;
+import com.google.common.collect.ImmutableSet;
 import dagger.Module;
 import dagger.Provides;
 import google.registry.batch.ExpandRecurringBillingEventsAction;
@@ -39,6 +42,14 @@ public class BackendModule {
     return assertTldExists(extractRequiredParameter(req, RequestParameters.PARAM_TLD));
   }
 
+  @Provides
+  @Parameter(RequestParameters.PARAM_TLD)
+  static ImmutableSet<String> provideTlds(HttpServletRequest req) {
+    ImmutableSet<String> tlds = extractSetOfParameters(req, RequestParameters.PARAM_TLD);
+    assertTldsExist(tlds);
+    return tlds;
+  }
+
   @Provides
   @Parameter("cursorTime")
   static Optional<DateTime> provideCursorTime(HttpServletRequest req) {

java/google/registry/rde/PendingDeposit.java

@@ -15,6 +15,7 @@
 package google.registry.rde;
 
 import com.google.auto.value.AutoValue;
+import com.google.common.base.Optional;
 import google.registry.model.common.Cursor.CursorType;
 import google.registry.model.rde.RdeMode;
 import java.io.Serializable;
@@ -27,15 +28,67 @@ public abstract class PendingDeposit implements Serializable {
 
   private static final long serialVersionUID = 3141095605225904433L;
 
+  /**
+   * True if deposits should be generated via manual operation, which does not update the cursor,
+   * and saves the generated deposits in a special manual subdirectory tree.
+   */
+  public abstract boolean manual();
+
+  /** TLD for which a deposit should be generated. */
   public abstract String tld();
+
+  /** Watermark date for which a deposit should be generated. */
   public abstract DateTime watermark();
+
+  /** Which type of deposit to generate: full (RDE) or thin (BRDA). */
   public abstract RdeMode mode();
-  public abstract CursorType cursor();
-  public abstract Duration interval();
+
+  /** The cursor type to update (not used in manual operation). */
+  public abstract Optional<CursorType> cursor();
+
+  /** Amount of time to increment the cursor (not used in manual operation). */
+  public abstract Optional<Duration> interval();
+
+  /**
+   * Subdirectory of bucket/manual in which files should be placed, including a trailing slash (used
+   * only in manual operation).
+   */
+  public abstract Optional<String> directoryWithTrailingSlash();
+
+  /**
+   * Revision number for generated files; if absent, use the next available in the sequence (used
+   * only in manual operation).
+   */
+  public abstract Optional<Integer> revision();
 
   static PendingDeposit create(
       String tld, DateTime watermark, RdeMode mode, CursorType cursor, Duration interval) {
-    return new AutoValue_PendingDeposit(tld, watermark, mode, cursor, interval);
+    return new AutoValue_PendingDeposit(
+        false,
+        tld,
+        watermark,
+        mode,
+        Optional.of(cursor),
+        Optional.of(interval),
+        Optional.<String>absent(),
+        Optional.<Integer>absent());
+  }
+
+  static PendingDeposit createInManualOperation(
+      String tld,
+      DateTime watermark,
+      RdeMode mode,
+      String directoryWithTrailingSlash,
+      Optional<Integer> revision) {
+    return new AutoValue_PendingDeposit(
+        true,
+        tld,
+        watermark,
+        mode,
+        Optional.<CursorType>absent(),
+        Optional.<Duration>absent(),
+        Optional.of(directoryWithTrailingSlash),
+        revision);
   }
 
   PendingDeposit() {}

java/google/registry/rde/RdeModule.java

@@ -15,12 +15,19 @@
 package google.registry.rde;
 
 import static com.google.appengine.api.taskqueue.QueueFactory.getQueue;
+import static google.registry.request.RequestParameters.extractBooleanParameter;
+import static google.registry.request.RequestParameters.extractOptionalIntParameter;
+import static google.registry.request.RequestParameters.extractOptionalParameter;
+import static google.registry.request.RequestParameters.extractRequiredDatetimeParameter;
+import static google.registry.request.RequestParameters.extractSetOfDatetimeParameters;
+import static google.registry.request.RequestParameters.extractSetOfParameters;
 
 import com.google.appengine.api.taskqueue.Queue;
+import com.google.common.base.Optional;
+import com.google.common.collect.ImmutableSet;
 import dagger.Module;
 import dagger.Provides;
 import google.registry.request.Parameter;
-import google.registry.request.RequestParameters;
 import javax.inject.Named;
 import javax.servlet.http.HttpServletRequest;
 import org.joda.time.DateTime;
@@ -34,12 +41,45 @@ import org.joda.time.DateTime;
 public final class RdeModule {
 
   static final String PARAM_WATERMARK = "watermark";
-  static final String PATH = "path";
+  static final String PARAM_MANUAL = "manual";
+  static final String PARAM_DIRECTORY = "directory";
+  static final String PARAM_MODE = "mode";
+  static final String PARAM_REVISION = "revision";
 
   @Provides
   @Parameter(PARAM_WATERMARK)
   static DateTime provideWatermark(HttpServletRequest req) {
-    return DateTime.parse(RequestParameters.extractRequiredParameter(req, PARAM_WATERMARK));
+    return extractRequiredDatetimeParameter(req, PARAM_WATERMARK);
+  }
+
+  @Provides
+  @Parameter(PARAM_WATERMARK)
+  static ImmutableSet<DateTime> provideWatermarks(HttpServletRequest req) {
+    return extractSetOfDatetimeParameters(req, PARAM_WATERMARK);
+  }
+
+  @Provides
+  @Parameter(PARAM_MANUAL)
+  static boolean provideManual(HttpServletRequest req) {
+    return extractBooleanParameter(req, PARAM_MANUAL);
+  }
+
+  @Provides
+  @Parameter(PARAM_DIRECTORY)
+  static Optional<String> provideDirectory(HttpServletRequest req) {
+    return extractOptionalParameter(req, PARAM_DIRECTORY);
+  }
+
+  @Provides
+  @Parameter(PARAM_MODE)
+  static ImmutableSet<String> provideMode(HttpServletRequest req) {
+    return extractSetOfParameters(req, PARAM_MODE);
+  }
+
+  @Provides
+  @Parameter(PARAM_REVISION)
+  static Optional<Integer> provideRevision(HttpServletRequest req) {
+    return extractOptionalIntParameter(req, PARAM_REVISION);
   }
 
   @Provides

java/google/registry/rde/RdeStagingAction.java

@@ -17,8 +17,11 @@ package google.registry.rde;
 import static google.registry.util.PipelineUtils.createJobPath;
 import static javax.servlet.http.HttpServletResponse.SC_NO_CONTENT;
 
+import com.google.common.base.Ascii;
+import com.google.common.base.Optional;
 import com.google.common.base.Predicate;
 import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.ImmutableSetMultimap;
 import com.google.common.collect.Multimaps;
 import google.registry.config.RegistryConfig.Config;
@@ -34,10 +37,14 @@ import google.registry.model.index.EppResourceIndex;
 import google.registry.model.rde.RdeMode;
 import google.registry.model.registrar.Registrar;
 import google.registry.request.Action;
+import google.registry.request.HttpException.BadRequestException;
+import google.registry.request.Parameter;
+import google.registry.request.RequestParameters;
 import google.registry.request.Response;
 import google.registry.util.Clock;
 import google.registry.util.FormattingLogger;
 import javax.inject.Inject;
+import org.joda.time.DateTime;
 import org.joda.time.Duration;
 
 /**
@@ -47,7 +54,7 @@ import org.joda.time.Duration;
  *
  * <p>This task starts by asking {@link PendingDepositChecker} which deposits need to be generated.
  * If there's nothing to deposit, we return 204 No Content; otherwise, we fire off a MapReduce job
- * and redirect to its status GUI.
+ * and redirect to its status GUI. The task can also be run in manual operation, as described below.
  *
  * <p>The mapreduce job scans every {@link EppResource} in Datastore. It maps a point-in-time
  * representation of each entity to the escrow XML files in which it should appear.
@@ -150,6 +157,27 @@ import org.joda.time.Duration;
  *   guarantee referential correctness of your deposits, you must never delete a registrar entity.
  * </ul>
  *
+ * <h3>Manual Operation</h3>
+ *
+ * <p>The task can be run in manual operation by setting certain parameters. Rather than generating
+ * deposits which are currently outstanding, the task will generate specific deposits. The files
+ * will be stored in a subdirectory of the "manual" directory, to avoid overwriting regular deposit
+ * files. Cursors and revision numbers will not be updated, and the upload task will not be kicked
+ * off. The parameters are:
+ * <ul>
+ * <li>manual: if present and true, manual operation is indicated
+ * <li>directory: the subdirectory of "manual" into which the files should be placed
+ * <li>mode: the mode(s) to generate: FULL for RDE deposits, THIN for BRDA deposits
+ * <li>tld: the tld(s) for which deposits should be generated
+ * <li>watermark: the date(s) for which deposits should be generated; dates should be start-of-day
+ * <li>revision: optional; if not specified, the next available revision number will be used
+ * </ul>
+ *
+ * <p>The manual, directory, mode, tld and watermark parameters must be present for manual
+ * operation; they must all be absent for standard operation (except that manual can be present but
+ * set to false). The revision parameter is optional in manual operation, and must be absent for
+ * standard operation.
+ *
  * @see <a href="https://tools.ietf.org/html/draft-arias-noguchi-registry-data-escrow-06">Registry Data Escrow Specification</a>
  * @see <a href="https://tools.ietf.org/html/draft-arias-noguchi-dnrd-objects-mapping-05">Domain Name Registration Data Objects Mapping</a>
  */
@@ -164,23 +192,20 @@ public final class RdeStagingAction implements Runnable {
   @Inject Response response;
   @Inject MapreduceRunner mrRunner;
   @Inject @Config("transactionCooldown") Duration transactionCooldown;
+  @Inject @Parameter(RdeModule.PARAM_MANUAL) boolean manual;
+  @Inject @Parameter(RdeModule.PARAM_DIRECTORY) Optional<String> directory;
+  @Inject @Parameter(RdeModule.PARAM_MODE) ImmutableSet<String> modeStrings;
+  @Inject @Parameter(RequestParameters.PARAM_TLD) ImmutableSet<String> tlds;
+  @Inject @Parameter(RdeModule.PARAM_WATERMARK) ImmutableSet<DateTime> watermarks;
+  @Inject @Parameter(RdeModule.PARAM_REVISION) Optional<Integer> revision;
+
   @Inject RdeStagingAction() {}
 
   @Override
   public void run() {
-    ImmutableSetMultimap<String, PendingDeposit> pendings = ImmutableSetMultimap.copyOf(
-        Multimaps.filterValues(
-            pendingDepositChecker.getTldsAndWatermarksPendingDepositForRdeAndBrda(),
-            new Predicate<PendingDeposit>() {
-              @Override
-              public boolean apply(PendingDeposit pending) {
-                if (clock.nowUtc().isBefore(pending.watermark().plus(transactionCooldown))) {
-                  logger.infofmt("Ignoring within %s cooldown: %s", transactionCooldown, pending);
-                  return false;
-                } else {
-                  return true;
-                }
-              }}));
+    ImmutableSetMultimap<String, PendingDeposit> pendings =
+        manual ? getManualPendingDeposits() : getStandardPendingDeposits();
+
     if (pendings.isEmpty()) {
       String message = "Nothing needs to be deposited";
       logger.info(message);
@@ -188,6 +213,7 @@ public final class RdeStagingAction implements Runnable {
       response.setPayload(message);
       return;
     }
+
     for (PendingDeposit pending : pendings.values()) {
       logger.infofmt("%s", pending);
     }
@@ -203,4 +229,97 @@ public final class RdeStagingAction implements Runnable {
                 new NullInput<EppResource>(),
                 EppResourceInputs.createEntityInput(EppResource.class)))));
   }
+
+  private ImmutableSetMultimap<String, PendingDeposit> getStandardPendingDeposits() {
+    if (directory.isPresent()) {
+      throw new BadRequestException("Directory parameter not allowed in standard operation");
+    }
+    if (!modeStrings.isEmpty()) {
+      throw new BadRequestException("Mode parameter not allowed in standard operation");
+    }
+    if (!tlds.isEmpty()) {
+      throw new BadRequestException("Tld parameter not allowed in standard operation");
+    }
+    if (!watermarks.isEmpty()) {
+      throw new BadRequestException("Watermark parameter not allowed in standard operation");
+    }
+    if (revision.isPresent()) {
+      throw new BadRequestException("Revision parameter not allowed in standard operation");
+    }
+
+    return ImmutableSetMultimap.copyOf(
+        Multimaps.filterValues(
+            pendingDepositChecker.getTldsAndWatermarksPendingDepositForRdeAndBrda(),
+            new Predicate<PendingDeposit>() {
+              @Override
+              public boolean apply(PendingDeposit pending) {
+                if (clock.nowUtc().isBefore(pending.watermark().plus(transactionCooldown))) {
+                  logger.infofmt("Ignoring within %s cooldown: %s", transactionCooldown, pending);
+                  return false;
+                } else {
+                  return true;
+                }
+              }}));
+  }
+
+  private ImmutableSetMultimap<String, PendingDeposit> getManualPendingDeposits() {
+    if (!directory.isPresent()) {
+      throw new BadRequestException("Directory parameter required in manual operation");
+    }
+    if (directory.get().startsWith("/")) {
+      throw new BadRequestException("Directory must not start with a slash");
+    }
+    String directoryWithTrailingSlash =
+        directory.get().endsWith("/") ? directory.get() : (directory.get() + '/');
+
+    if (modeStrings.isEmpty()) {
+      throw new BadRequestException("Mode parameter required in manual operation");
+    }
+
+    ImmutableSet.Builder<RdeMode> modesBuilder = new ImmutableSet.Builder<>();
+    for (String modeString : modeStrings) {
+      try {
+        modesBuilder.add(RdeMode.valueOf(Ascii.toUpperCase(modeString)));
+      } catch (IllegalArgumentException e) {
+        throw new BadRequestException("Mode must be FULL for RDE deposits, THIN for BRDA deposits");
+      }
+    }
+    ImmutableSet<RdeMode> modes = modesBuilder.build();
+
+    if (tlds.isEmpty()) {
+      throw new BadRequestException("Tld parameter required in manual operation");
+    }
+
+    if (watermarks.isEmpty()) {
+      throw new BadRequestException("Watermark parameter required in manual operation");
+    }
+    // In theory, BRDA deposits should be on a specific day of the week, but in manual mode, let the
+    // user create deposits on other days. But dates should definitely be at the start of the day;
+    // otherwise, confusion is likely.
+    for (DateTime watermark : watermarks) {
+      if (!watermark.equals(watermark.withTimeAtStartOfDay())) {
+        throw new BadRequestException("Watermarks must be at the start of a day.");
+      }
+    }
+
+    ImmutableSetMultimap.Builder<String, PendingDeposit> pendingsBuilder =
+        new ImmutableSetMultimap.Builder<>();
+
+    for (String tld : tlds) {
+      for (DateTime watermark : watermarks) {
+        for (RdeMode mode : modes) {
+          pendingsBuilder.put(
+              tld,
+              PendingDeposit.createInManualOperation(
+                  tld,
+                  watermark,
+                  mode,
+                  directoryWithTrailingSlash,
+                  revision));
+        }
+      }
+    }
+
+    return pendingsBuilder.build();
+  }
 }

java/google/registry/rde/RdeStagingReducer.java

@@ -17,6 +17,7 @@ package google.registry.rde;
 import static com.google.appengine.api.taskqueue.QueueFactory.getQueue;
 import static com.google.appengine.api.taskqueue.TaskOptions.Builder.withUrl;
 import static com.google.appengine.tools.cloudstorage.GcsServiceFactory.createGcsService;
+import static com.google.common.base.Preconditions.checkState;
 import static com.google.common.base.Verify.verify;
 import static google.registry.model.common.Cursor.getCursorTimeOrStartOfTime;
 import static google.registry.model.ofy.ObjectifyService.ofy;
@@ -107,9 +108,13 @@ public final class RdeStagingReducer extends Reducer<PendingDeposit, DepositFrag
     final RdeMode mode = key.mode();
     final String tld = key.tld();
     final DateTime watermark = key.watermark();
-    final int revision = RdeRevision.getNextRevision(tld, watermark, mode);
+    final int revision = key.revision().or(RdeRevision.getNextRevision(tld, watermark, mode));
     String id = RdeUtil.timestampToId(watermark);
     String prefix = RdeNamingUtils.makeRydeFilename(tld, watermark, mode, 1, revision);
+    if (key.manual()) {
+      checkState(key.directoryWithTrailingSlash().isPresent(), "Manual subdirectory not specified");
+      prefix = "manual/" + key.directoryWithTrailingSlash().get() + prefix;
+    }
     GcsFilename xmlFilename = new GcsFilename(bucket, prefix + ".xml.ghostryde");
     GcsFilename xmlLengthFilename = new GcsFilename(bucket, prefix + ".xml.length");
     GcsFilename reportFilename = new GcsFilename(bucket, prefix + "-report.xml.ghostryde");
@@ -193,20 +198,25 @@ public final class RdeStagingReducer extends Reducer<PendingDeposit, DepositFrag
     }
 
     // Now that we're done, kick off RdeUploadAction and roll forward the cursor transactionally.
+    if (key.manual()) {
+      logger.info("Manual operation; not advancing cursor or enqueuing upload task");
+      return;
+    }
     ofy().transact(new VoidWork() {
       @Override
       public void vrun() {
         Registry registry = Registry.get(tld);
         DateTime position = getCursorTimeOrStartOfTime(
-            ofy().load().key(Cursor.createKey(key.cursor(), registry)).now());
-        DateTime newPosition = key.watermark().plus(key.interval());
+            ofy().load().key(Cursor.createKey(key.cursor().get(), registry)).now());
+        checkState(key.interval().isPresent(), "Interval must be present");
+        DateTime newPosition = key.watermark().plus(key.interval().get());
         if (!position.isBefore(newPosition)) {
           logger.warning("Cursor has already been rolled forward.");
           return;
         }
         verify(position.equals(key.watermark()),
             "Partial ordering of RDE deposits broken: %s %s", position, key);
-        ofy().save().entity(Cursor.create(key.cursor(), newPosition, registry)).now();
+        ofy().save().entity(Cursor.create(key.cursor().get(), newPosition, registry)).now();
         logger.infofmt("Rolled forward %s on %s cursor to %s", key.cursor(), tld, newPosition);
         RdeRevision.saveRevision(tld, watermark, mode, revision);
         if (mode == RdeMode.FULL) {

java/google/registry/request/RequestParameters.java

@@ -175,6 +175,33 @@ public final class RequestParameters {
     }
   }
 
+  /**
+   * Returns all GET or POST date parameters associated with {@code name}, or an empty set if none.
+   *
+   * <p>Dates are parsed as an <a href="https://goo.gl/pk5Q2k">ISO 8601</a> timestamp, e.g. {@code
+   * 1984-12-18TZ}, {@code 2000-01-01T16:20:00Z}.
+   *
+   * @throws BadRequestException if one of the parameter values is not a valid {@link DateTime}.
+   */
+  public static ImmutableSet<DateTime> extractSetOfDatetimeParameters(
+      HttpServletRequest req, String name) {
+    String[] stringParams = req.getParameterValues(name);
+    if (stringParams == null) {
+      return ImmutableSet.<DateTime>of();
+    }
+    ImmutableSet.Builder<DateTime> datesBuilder = new ImmutableSet.Builder<>();
+    for (String stringParam : stringParams) {
+      try {
+        if (!isNullOrEmpty(stringParam)) {
+          datesBuilder.add(DateTime.parse(stringParam));
+        }
+      } catch (IllegalArgumentException e) {
+        throw new BadRequestException("Bad ISO 8601 timestamp: " + name);
+      }
+    }
+    return datesBuilder.build();
+  }
+
   /**
    * Returns first request parameter associated with {@code name} parsed as an optional
    * {@link InetAddress} (which might be IPv6).

java/google/registry/tools/GenerateEscrowDepositCommand.java

@@ -81,7 +81,7 @@ final class GenerateEscrowDepositCommand implements RemoteApiCommand {
 
   @Parameter(
       names = {"-m", "--mode"},
-      description = "RDE/BRDA mode of operation.")
+      description = "FULL/THIN mode of operation.")
   private RdeMode mode = RdeMode.FULL;
 
   @Parameter(

javatests/google/registry/rde/RdeStagingActionTest.java

@@ -36,6 +36,13 @@ import static java.util.Arrays.asList;
 import com.google.appengine.tools.cloudstorage.GcsFilename;
 import com.google.appengine.tools.cloudstorage.GcsService;
 import com.google.appengine.tools.cloudstorage.GcsServiceFactory;
+import com.google.appengine.tools.cloudstorage.ListItem;
+import com.google.appengine.tools.cloudstorage.ListOptions;
+import com.google.appengine.tools.cloudstorage.ListResult;
+import com.google.common.base.Function;
+import com.google.common.base.Optional;
+import com.google.common.collect.FluentIterable;
+import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
 import com.google.common.net.InetAddresses;
@@ -47,7 +54,9 @@ import google.registry.model.common.Cursor.CursorType;
 import google.registry.model.host.HostResource;
 import google.registry.model.ofy.Ofy;
 import google.registry.model.registry.Registry;
+import google.registry.request.HttpException.BadRequestException;
 import google.registry.request.RequestParameters;
+import google.registry.testing.ExceptionRule;
 import google.registry.testing.FakeClock;
 import google.registry.testing.FakeKeyringModule;
 import google.registry.testing.FakeResponse;
@@ -101,6 +110,9 @@ public class RdeStagingActionTest extends MapreduceTestCase<RdeStagingAction> {
   @Rule
   public final InjectRule inject = new InjectRule();
 
+  @Rule
+  public final ExceptionRule thrown = new ExceptionRule();
+
   private final FakeClock clock = new FakeClock();
   private final FakeResponse response = new FakeResponse();
   private final GcsService gcsService = GcsServiceFactory.createGcsService();
@@ -136,6 +148,47 @@ public class RdeStagingActionTest extends MapreduceTestCase<RdeStagingAction> {
     action.pendingDepositChecker.rdeInterval = Duration.standardDays(1);
     action.response = response;
     action.transactionCooldown = Duration.ZERO;
+    action.directory = Optional.<String>absent();
+    action.modeStrings = ImmutableSet.<String>of();
+    action.tlds = ImmutableSet.<String>of();
+    action.watermarks = ImmutableSet.<DateTime>of();
+    action.revision = Optional.<Integer>absent();
+  }
+
+  @Test
+  public void testRun_modeInNonManualMode_throwsException() throws Exception {
+    createTldWithEscrowEnabled("lol");
+    clock.setTo(DateTime.parse("2000-01-01TZ"));
+    action.modeStrings = ImmutableSet.of("full");
+    thrown.expect(BadRequestException.class);
+    action.run();
+  }
+
+  @Test
+  public void testRun_tldInNonManualMode_throwsException() throws Exception {
+    createTldWithEscrowEnabled("lol");
+    clock.setTo(DateTime.parse("2000-01-01TZ"));
+    action.tlds = ImmutableSet.of("tld");
+    thrown.expect(BadRequestException.class);
+    action.run();
+  }
+
+  @Test
+  public void testRun_watermarkInNonManualMode_throwsException() throws Exception {
+    createTldWithEscrowEnabled("lol");
+    clock.setTo(DateTime.parse("2000-01-01TZ"));
+    action.watermarks = ImmutableSet.of(clock.nowUtc());
+    thrown.expect(BadRequestException.class);
+    action.run();
+  }
+
+  @Test
+  public void testRun_revisionInNonManualMode_throwsException() throws Exception {
+    createTldWithEscrowEnabled("lol");
+    clock.setTo(DateTime.parse("2000-01-01TZ"));
+    action.revision = Optional.of(42);
+    thrown.expect(BadRequestException.class);
+    action.run();
   }
 
   @Test
@@ -184,6 +237,86 @@ public class RdeStagingActionTest extends MapreduceTestCase<RdeStagingAction> {
     assertAtLeastOneTaskIsEnqueued("mapreduce");
   }
 
+  @Test
+  public void testManualRun_emptyMode_throwsException() throws Exception {
+    createTldWithEscrowEnabled("lol");
+    clock.setTo(DateTime.parse("2000-01-01TZ"));
+    action.manual = true;
+    action.directory = Optional.of("test/");
+    action.modeStrings = ImmutableSet.<String>of();
+    action.tlds = ImmutableSet.of("lol");
+    action.watermarks = ImmutableSet.of(clock.nowUtc());
+    thrown.expect(BadRequestException.class);
+    action.run();
+  }
+
+  @Test
+  public void testManualRun_invalidMode_throwsException() throws Exception {
+    createTldWithEscrowEnabled("lol");
+    clock.setTo(DateTime.parse("2000-01-01TZ"));
+    action.manual = true;
+    action.directory = Optional.of("test/");
+    action.modeStrings = ImmutableSet.of("full", "thing");
+    action.tlds = ImmutableSet.of("lol");
+    action.watermarks = ImmutableSet.of(clock.nowUtc());
+    thrown.expect(BadRequestException.class);
+    action.run();
+  }
+
+  @Test
+  public void testManualRun_emptyTld_throwsException() throws Exception {
+    createTldWithEscrowEnabled("lol");
+    clock.setTo(DateTime.parse("2000-01-01TZ"));
+    action.manual = true;
+    action.directory = Optional.of("test/");
+    action.modeStrings = ImmutableSet.of("full");
+    action.tlds = ImmutableSet.<String>of();
+    action.watermarks = ImmutableSet.of(clock.nowUtc());
+    thrown.expect(BadRequestException.class);
+    action.run();
+  }
+
+  @Test
+  public void testManualRun_emptyWatermark_throwsException() throws Exception {
+    createTldWithEscrowEnabled("lol");
+    clock.setTo(DateTime.parse("2000-01-01TZ"));
+    action.manual = true;
+    action.directory = Optional.of("test/");
+    action.modeStrings = ImmutableSet.of("full");
+    action.tlds = ImmutableSet.of("lol");
+    action.watermarks = ImmutableSet.of();
+    thrown.expect(BadRequestException.class);
+    action.run();
+  }
+
+  @Test
+  public void testManualRun_nonDayStartWatermark_throwsException() throws Exception {
+    createTldWithEscrowEnabled("lol");
+    clock.setTo(DateTime.parse("2000-01-01TZ"));
+    action.manual = true;
+    action.directory = Optional.of("test/");
+    action.modeStrings = ImmutableSet.of("full");
+    action.tlds = ImmutableSet.of("lol");
+    action.watermarks = ImmutableSet.of(DateTime.parse("2001-01-01T01:36:45Z"));
+    thrown.expect(BadRequestException.class);
+    action.run();
+  }
+
+  @Test
+  public void testManualRun_validParameters_runsMapReduce() throws Exception {
+    createTldWithEscrowEnabled("lol");
+    clock.setTo(DateTime.parse("2000-01-01TZ"));
+    action.manual = true;
+    action.directory = Optional.of("test/");
+    action.modeStrings = ImmutableSet.of("full");
+    action.tlds = ImmutableSet.of("lol");
+    action.watermarks = ImmutableSet.of(DateTime.parse("2001-01-01TZ"));
+    action.run();
+    assertThat(response.getStatus()).isEqualTo(200);
+    assertThat(response.getPayload()).contains("_ah/pipeline/status.html?root=");
+    assertAtLeastOneTaskIsEnqueued("mapreduce");
+  }
+
   @Test
   public void testMapReduce_bunchOfResources_headerHasCorrectCounts() throws Exception {
     clock.setTo(DateTime.parse("1999-12-31TZ"));
@@ -610,6 +743,87 @@ public class RdeStagingActionTest extends MapreduceTestCase<RdeStagingAction> {
         .isEqualTo(DateTime.parse("1984-12-21TZ"));
   }
 
+  private void doManualModeMapReduceTest(int revision, ImmutableSet<String> tlds) throws Exception {
+    clock.setTo(DateTime.parse("1999-12-31TZ"));
+    for (String tld : tlds) {
+      createTldWithEscrowEnabled(tld);
+      makeDomainResource(clock, tld);
+      setCursor(Registry.get(tld), RDE_STAGING, DateTime.parse("1999-01-01TZ"));
+      setCursor(Registry.get(tld), BRDA, DateTime.parse("2001-01-01TZ"));
+    }
+
+    action.manual = true;
+    action.directory = Optional.of("test/");
+    action.modeStrings = ImmutableSet.of("full", "thin");
+    action.tlds = tlds;
+    action.watermarks =
+        ImmutableSet.of(DateTime.parse("2000-01-01TZ"), DateTime.parse("2000-01-02TZ"));
+    action.revision = Optional.of(revision);
+
+    action.run();
+    executeTasksUntilEmpty("mapreduce", clock);
+
+    ListResult listResult =
+        gcsService.list("rde-bucket", new ListOptions.Builder().setPrefix("manual/test").build());
+    ImmutableSet<String> filenames =
+        FluentIterable.from(ImmutableList.copyOf(listResult))
+            .transform(
+                new Function<ListItem, String>() {
+                  @Override
+                  public String apply(ListItem listItem) {
+                    return listItem.getName();
+                  }
+                })
+            .toSet();
+    for (String tld : tlds) {
+      assertThat(filenames)
+          .containsAllOf(
+              "manual/test/" + tld + "_2000-01-01_full_S1_R" + revision + "-report.xml.ghostryde",
+              "manual/test/" + tld + "_2000-01-01_full_S1_R" + revision + ".xml.ghostryde",
+              "manual/test/" + tld + "_2000-01-01_full_S1_R" + revision + ".xml.length",
+              "manual/test/" + tld + "_2000-01-01_thin_S1_R" + revision + ".xml.ghostryde",
+              "manual/test/" + tld + "_2000-01-01_thin_S1_R" + revision + ".xml.length",
+              "manual/test/" + tld + "_2000-01-02_full_S1_R" + revision + "-report.xml.ghostryde",
+              "manual/test/" + tld + "_2000-01-02_full_S1_R" + revision + ".xml.ghostryde",
+              "manual/test/" + tld + "_2000-01-02_full_S1_R" + revision + ".xml.length",
+              "manual/test/" + tld + "_2000-01-02_thin_S1_R" + revision + ".xml.ghostryde",
+              "manual/test/" + tld + "_2000-01-02_thin_S1_R" + revision + ".xml.length");
+
+      assertThat(
+              ofy()
+                  .load()
+                  .key(Cursor.createKey(RDE_STAGING, Registry.get(tld)))
+                  .now()
+                  .getCursorTime())
+          .isEqualTo(DateTime.parse("1999-01-01TZ"));
+      assertThat(ofy().load().key(Cursor.createKey(BRDA, Registry.get(tld))).now().getCursorTime())
+          .isEqualTo(DateTime.parse("2001-01-01TZ"));
+    }
+  }
+
+  @Test
+  public void testMapReduce_manualMode_generatesCorrectDepositsWithoutAdvancingCursors()
+      throws Exception {
+    doManualModeMapReduceTest(0, ImmutableSet.of("lol"));
+    XmlTestUtils.assertXmlEquals(
+        readResourceUtf8(getClass(), "testdata/testMapReduce_withDomain_producesExpectedXml.xml"),
+        readXml("manual/test/lol_2000-01-01_full_S1_R0.xml.ghostryde"),
+        "deposit.contents.registrar.crDate",
+        "deposit.contents.registrar.upDate");
+    XmlTestUtils.assertXmlEquals(
+        readResourceUtf8(getClass(), "testdata/testMapReduce_withDomain_producesReportXml.xml"),
+        readXml(
+            "manual/test/lol_2000-01-01_full_S1_R0-report.xml.ghostryde"),
+        "deposit.contents.registrar.crDate",
+        "deposit.contents.registrar.upDate");
+  }
+
+  @Test
+  public void testMapReduce_manualMode_nonZeroRevisionAndMultipleTlds()
+      throws Exception {
+    doManualModeMapReduceTest(42, ImmutableSet.of("lol", "slug"));
+  }
+
   private String readXml(String objectName) throws IOException, PGPException {
     GcsFilename file = new GcsFilename("rde-bucket", objectName);
     return new String(Ghostryde.decode(readGcsFile(gcsService, file), decryptKey).getData(), UTF_8);