Remove static methods in back up actions (#1481)

* Remove static methods in back up actions

* Remove BigqueryPollJob helper class

* Add schedule time in task comparison

* Change payload type from byte[] to ByteString

core/src/main/java/google/registry/backup/CommitLogCheckpointAction.java

@@ -28,11 +28,11 @@ import google.registry.model.ofy.CommitLogCheckpointRoot;
 import google.registry.request.Action;
 import google.registry.request.Action.Service;
 import google.registry.request.auth.Auth;
-import google.registry.util.Clock;
 import google.registry.util.CloudTasksUtils;
 import java.util.Optional;
 import javax.inject.Inject;
 import org.joda.time.DateTime;
+import org.joda.time.Duration;
 
 /**
  * Action that saves commit log checkpoints to Datastore and kicks off a diff export task.
@@ -57,7 +57,22 @@ public final class CommitLogCheckpointAction implements Runnable {
 
   private static final String QUEUE_NAME = "export-commits";
 
-  @Inject Clock clock;
+  /**
+   * The amount of time enqueueing should be delayed.
+   *
+   * <p>The {@link ExportCommitLogDiffAction} is enqueued in {@link CommitLogCheckpointAction},
+   * which is inside a Datastore transaction that persists the checkpoint to be exported. After the
+   * switch to CloudTasks API, the task may be invoked before the Datastore transaction commits.
+   * When this happens, the checkpoint is not found which leads to {@link
+   * com.google.common.base.VerifyException}.
+   *
+   * <p>In order to invoke the task after the transaction commits, a reasonable delay should be
+   * added to each task. The latency of the request is mostly in the range of 4-6 seconds; Choosing
+   * a value 30% greater than the upper bound should solve the issue invoking a task before the
+   * transaction commits.
+   */
+  static final Duration ENQUEUE_DELAY_SECONDS = Duration.standardSeconds(8);
+
   @Inject CommitLogCheckpointStrategy strategy;
   @Inject CloudTasksUtils cloudTasksUtils;
 
@@ -96,14 +111,15 @@ public final class CommitLogCheckpointAction implements Runnable {
                   // Enqueue a diff task between previous and current checkpoints.
                   cloudTasksUtils.enqueue(
                       QUEUE_NAME,
-                      cloudTasksUtils.createPostTask(
+                      cloudTasksUtils.createPostTaskWithDelay(
                           ExportCommitLogDiffAction.PATH,
                           Service.BACKEND.toString(),
                           ImmutableMultimap.of(
                               LOWER_CHECKPOINT_TIME_PARAM,
                               lastWrittenTime.toString(),
                               UPPER_CHECKPOINT_TIME_PARAM,
-                              checkpoint.getCheckpointTime().toString())));
+                              checkpoint.getCheckpointTime().toString()),
+                          ENQUEUE_DELAY_SECONDS));
                   return true;
                 });
     return isCheckPointPersisted ? Optional.of(checkpoint) : Optional.empty();

core/src/main/java/google/registry/beam/rde/RdeIO.java

@@ -68,6 +68,7 @@ import org.apache.beam.sdk.values.PDone;
 import org.bouncycastle.jce.provider.BouncyCastleProvider;
 import org.bouncycastle.openpgp.PGPPublicKey;
 import org.joda.time.DateTime;
+import org.joda.time.Duration;
 
 public class RdeIO {
 
@@ -261,6 +262,7 @@ public class RdeIO {
 
     private static final FluentLogger logger = FluentLogger.forEnclosingClass();
     private static final long serialVersionUID = 5822176227753327224L;
+    private static final Duration ENQUEUE_DELAY = Duration.standardMinutes(1);
 
     private final CloudTasksUtils cloudTasksUtils;
 
@@ -271,6 +273,7 @@ public class RdeIO {
     @ProcessElement
     public void processElement(
         @Element KV<PendingDeposit, Integer> input, PipelineOptions options) {
+
       tm().transact(
               () -> {
                 PendingDeposit key = input.getKey();
@@ -296,26 +299,30 @@ public class RdeIO {
                 tm().put(Cursor.create(key.cursor(), newPosition, registry));
                 logger.atInfo().log(
                     "Rolled forward %s on %s cursor to %s.", key.cursor(), key.tld(), newPosition);
-                RdeRevision.saveRevision(key.tld(), key.watermark(), key.mode(), revision);
+                RdeRevision.saveRevision(key.tld(), key.watermark(), key.mode(), input.getValue());
                 // Enqueueing a task is a side effect that is not undone if the transaction rolls
                 // back. So this may result in multiple copies of the same task being processed.
                 // This is fine because the RdeUploadAction is guarded by a lock and tracks progress
-                // by cursor. The BrdaCopyAction writes a file to GCS, which is an atomic action.
+                // by cursor. The BrdaCopyAction writes a file to GCS, which is an atomic action. It
+                // is also guarded by a cursor to not run before the cursor is updated. We also
+                // include a delay to minimize the chance that the enqueued job executes before the
+                // transaction is committed, which triggers a retry.
                 if (key.mode() == RdeMode.FULL) {
                   cloudTasksUtils.enqueue(
                       RDE_UPLOAD_QUEUE,
-                      cloudTasksUtils.createPostTask(
+                      cloudTasksUtils.createPostTaskWithDelay(
                           RdeUploadAction.PATH,
                           Service.BACKEND.getServiceId(),
                           ImmutableMultimap.of(
                               RequestParameters.PARAM_TLD,
                               key.tld(),
                               RdeModule.PARAM_PREFIX,
-                              options.getJobName() + '/')));
+                              options.getJobName() + '/'),
+                          ENQUEUE_DELAY));
                 } else {
                   cloudTasksUtils.enqueue(
                       BRDA_QUEUE,
-                      cloudTasksUtils.createPostTask(
+                      cloudTasksUtils.createPostTaskWithDelay(
                           BrdaCopyAction.PATH,
                           Service.BACKEND.getServiceId(),
                           ImmutableMultimap.of(
@@ -324,7 +331,8 @@ public class RdeIO {
                               RdeModule.PARAM_WATERMARK,
                               key.watermark().toString(),
                               RdeModule.PARAM_PREFIX,
-                              options.getJobName() + '/')));
+                              options.getJobName() + '/'),
+                          ENQUEUE_DELAY));
                 }
               });
     }

core/src/main/java/google/registry/config/files/default-config.yaml

@@ -486,9 +486,9 @@ sslCertificateValidation:
   # The minimum number of days between two successive expiring notification emails.
   expirationWarningIntervalDays: 15
   # Text for expiring certificate notification email subject.
-  expirationWarningEmailSubjectText: "[Important] Expiring SSL certificate for Google Registry EPP connection"
-  # Text for expiring certificate notification email body that accepts 3 parameters:
-  # registrar name, certificate type, and expiration date, respectively.
+  expirationWarningEmailSubjectText: "Expiring SSL certificate for Example Registry EPP connection"
+  # Text for expiring certificate notification email body that accepts 4 parameters:
+  # registrar name, certificate type, expiration date and registrar id, respectively.
   expirationWarningEmailBodyText: |
     Dear %1$s,
 
@@ -496,25 +496,15 @@ sslCertificateValidation:
 
     Kindly update your production account certificate within the support console using the following steps:
 
-      1. Navigate to support.registry.google and login using your %4$s@registry.google credentials.
-          * If this is your first time logging in, you will be prompted to reset your password, so please keep your new password safe.
-          * If you are already logged in with some other Google account(s) but not your %4$s@registry.google account, you need to click on
-          “Add Account” and login using your %4$s@registry.google credentials.
+      1. Navigate to support.registry.example and login using your %4$s@registry.example credentials.
       2. Select “Settings > Security” from the left navigation bar.
       3. Click “Edit” on the top left corner.
-      4. Enter your full certificate string (including lines -----BEGIN CERTIFICATE----- and -----END CERTIFICATE-----) in the box.
-      5. Click “Save”. If there are validation issues with the form, you will be prompted to fix them and click “Save” again.
+      4. Enter your full certificate string in the box.
+      5. Click “Save”.
 
-    A failover SSL certificate can also be added in order to prevent connection issues once your main certificate expires. Connecting with either of the certificates will work with our production EPP server.
-
-    Further information about our EPP connection requirements can be found in section 9.2 in the updated Technical Guide in your Google Drive folder.
-
-    Note that account certificate changes take a few minutes to become effective and that the existing connections will remain unaffected by the change.
-
-    If you also would like to update your OT&E account certificate, please send an email from your primary or technical contact to registry-support@google.com and include the full certificate string (including lines -----BEGIN CERTIFICATE----- and -----END CERTIFICATE-----).
 
     Regards,
-    Google Registry
+    Example Registry
 
   # The minimum number of bits an RSA key must contain.
   minimumRsaKeyLength: 2048

core/src/main/java/google/registry/export/BackupDatastoreAction.java

@@ -14,18 +14,22 @@
 
 package google.registry.export;
 
-import static google.registry.export.CheckBackupAction.enqueuePollTask;
 import static google.registry.request.Action.Method.POST;
 
+import com.google.common.base.Joiner;
+import com.google.common.collect.ImmutableMultimap;
 import com.google.common.flogger.FluentLogger;
 import google.registry.config.RegistryConfig;
 import google.registry.export.datastore.DatastoreAdmin;
 import google.registry.export.datastore.Operation;
 import google.registry.model.annotations.DeleteAfterMigration;
 import google.registry.request.Action;
+import google.registry.request.Action.Service;
 import google.registry.request.HttpException.InternalServerErrorException;
 import google.registry.request.Response;
 import google.registry.request.auth.Auth;
+import google.registry.util.Clock;
+import google.registry.util.CloudTasksUtils;
 import javax.inject.Inject;
 
 /**
@@ -59,6 +63,8 @@ public class BackupDatastoreAction implements Runnable {
 
   @Inject DatastoreAdmin datastoreAdmin;
   @Inject Response response;
+  @Inject Clock clock;
+  @Inject CloudTasksUtils cloudTasksUtils;
 
   @Inject
   BackupDatastoreAction() {}
@@ -73,8 +79,19 @@ public class BackupDatastoreAction implements Runnable {
               .execute();
 
       String backupName = backup.getName();
-      // Enqueue a poll task to monitor the backup and load REPORTING-related kinds into bigquery.
-      enqueuePollTask(backupName, AnnotatedEntities.getReportingKinds());
+      // Enqueue a poll task to monitor the backup for completion and load reporting-related kinds
+      // into bigquery.
+      cloudTasksUtils.enqueue(
+          CheckBackupAction.QUEUE,
+          cloudTasksUtils.createPostTaskWithDelay(
+              CheckBackupAction.PATH,
+              Service.BACKEND.toString(),
+              ImmutableMultimap.of(
+                  CheckBackupAction.CHECK_BACKUP_NAME_PARAM,
+                  backupName,
+                  CheckBackupAction.CHECK_BACKUP_KINDS_TO_LOAD_PARAM,
+                  Joiner.on(',').join(AnnotatedEntities.getReportingKinds())),
+              CheckBackupAction.POLL_COUNTDOWN));
       String message =
           String.format(
               "Datastore backup started with name: %s\nSaving to %s",

core/src/main/java/google/registry/export/BigqueryPollJobAction.java

@@ -14,18 +14,14 @@
 
 package google.registry.export;
 
-import static com.google.appengine.api.taskqueue.QueueFactory.getQueue;
-import static com.google.appengine.api.taskqueue.TaskOptions.Builder.withUrl;
 import static google.registry.bigquery.BigqueryUtils.toJobReferenceString;
 
 import com.google.api.services.bigquery.Bigquery;
 import com.google.api.services.bigquery.model.Job;
 import com.google.api.services.bigquery.model.JobReference;
-import com.google.appengine.api.taskqueue.Queue;
-import com.google.appengine.api.taskqueue.TaskHandle;
-import com.google.appengine.api.taskqueue.TaskOptions;
-import com.google.appengine.api.taskqueue.TaskOptions.Method;
+import com.google.cloud.tasks.v2.Task;
 import com.google.common.flogger.FluentLogger;
+import com.google.protobuf.ByteString;
 import dagger.Lazy;
 import google.registry.request.Action;
 import google.registry.request.Header;
@@ -33,12 +29,10 @@ import google.registry.request.HttpException.BadRequestException;
 import google.registry.request.HttpException.NotModifiedException;
 import google.registry.request.Payload;
 import google.registry.request.auth.Auth;
-import google.registry.util.TaskQueueUtils;
+import google.registry.util.CloudTasksUtils;
 import java.io.ByteArrayInputStream;
-import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.io.ObjectInputStream;
-import java.io.ObjectOutputStream;
 import javax.inject.Inject;
 import org.joda.time.Duration;
 
@@ -67,11 +61,14 @@ public class BigqueryPollJobAction implements Runnable {
   static final Duration POLL_COUNTDOWN = Duration.standardSeconds(20);
 
   @Inject Bigquery bigquery;
-  @Inject TaskQueueUtils taskQueueUtils;
+  @Inject CloudTasksUtils cloudTasksUtils;
+
   @Inject @Header(CHAINED_TASK_QUEUE_HEADER) Lazy<String> chainedQueueName;
   @Inject @Header(PROJECT_ID_HEADER) String projectId;
   @Inject @Header(JOB_ID_HEADER) String jobId;
-  @Inject @Payload byte[] payload;
+
+  @Inject @Payload ByteString payload;
+
   @Inject BigqueryPollJobAction() {}
 
   @Override
@@ -79,20 +76,25 @@ public class BigqueryPollJobAction implements Runnable {
     boolean jobOutcome =
         checkJobOutcome(); // Throws a NotModifiedException if the job hasn't completed.
     // If the job failed, do not enqueue the next step.
-    if (!jobOutcome || payload == null || payload.length == 0) {
+    if (!jobOutcome || payload == null || payload.size() == 0) {
       return;
     }
     // If there is a payload, it's a chained task, so enqueue it.
-    TaskOptions task;
+    Task task;
     try {
-      task = (TaskOptions) new ObjectInputStream(new ByteArrayInputStream(payload)).readObject();
+      task =
+          (Task)
+              new ObjectInputStream(new ByteArrayInputStream(payload.toByteArray())).readObject();
     } catch (ClassNotFoundException | IOException e) {
       throw new BadRequestException("Cannot deserialize task from payload", e);
     }
-    String taskName = taskQueueUtils.enqueue(getQueue(chainedQueueName.get()), task).getName();
+    Task enqueuedTask = cloudTasksUtils.enqueue(chainedQueueName.get(), task);
     logger.atInfo().log(
         "Added chained task %s for %s to queue %s: %s",
-        taskName, task.getUrl(), chainedQueueName.get(), task);
+        enqueuedTask.getName(),
+        enqueuedTask.getAppEngineHttpRequest().getRelativeUri(),
+        chainedQueueName.get(),
+        enqueuedTask);
   }
 
   /**
@@ -124,51 +126,4 @@ public class BigqueryPollJobAction implements Runnable {
     logger.atInfo().log("Bigquery job succeeded - %s.", jobRefString);
     return true;
   }
-
-
-  /** Helper class to enqueue a bigquery poll job. */
-  public static class BigqueryPollJobEnqueuer {
-
-    private final TaskQueueUtils taskQueueUtils;
-
-    @Inject
-    BigqueryPollJobEnqueuer(TaskQueueUtils taskQueueUtils) {
-      this.taskQueueUtils = taskQueueUtils;
-    }
-
-    /** Enqueue a task to poll for the success or failure of the referenced BigQuery job. */
-    public TaskHandle enqueuePollTask(JobReference jobRef) {
-      return taskQueueUtils.enqueue(
-          getQueue(QUEUE), createCommonPollTask(jobRef).method(Method.GET));
-    }
-
-    /**
-     * Enqueue a task to poll for the success or failure of the referenced BigQuery job and to
-     * launch the provided task in the specified queue if the job succeeds.
-     */
-    public TaskHandle enqueuePollTask(
-        JobReference jobRef, TaskOptions chainedTask, Queue chainedTaskQueue) throws IOException {
-      // Serialize the chainedTask into a byte array to put in the task payload.
-      ByteArrayOutputStream taskBytes = new ByteArrayOutputStream();
-      new ObjectOutputStream(taskBytes).writeObject(chainedTask);
-      return taskQueueUtils.enqueue(
-          getQueue(QUEUE),
-          createCommonPollTask(jobRef)
-              .method(Method.POST)
-              .header(CHAINED_TASK_QUEUE_HEADER, chainedTaskQueue.getQueueName())
-              .payload(taskBytes.toByteArray()));
-    }
-
-    /**
-     * Enqueue a task to poll for the success or failure of the referenced BigQuery job and to
-     * launch the provided task in the specified queue if the job succeeds.
-     */
-    private static TaskOptions createCommonPollTask(JobReference jobRef) {
-      // Omit host header so that task will be run on the current backend/module.
-      return withUrl(PATH)
-          .countdownMillis(POLL_COUNTDOWN.getMillis())
-          .header(PROJECT_ID_HEADER, jobRef.getProjectId())
-          .header(JOB_ID_HEADER, jobRef.getJobId());
-    }
-  }
 }

core/src/main/java/google/registry/export/CheckBackupAction.java

@@ -16,18 +16,14 @@ package google.registry.export;
 
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.collect.Sets.intersection;
-import static google.registry.export.UploadDatastoreBackupAction.enqueueUploadBackupTask;
 import static google.registry.request.Action.Method.GET;
 import static google.registry.request.Action.Method.POST;
 import static javax.servlet.http.HttpServletResponse.SC_NOT_FOUND;
 
 import com.google.api.client.googleapis.json.GoogleJsonResponseException;
-import com.google.appengine.api.taskqueue.QueueFactory;
-import com.google.appengine.api.taskqueue.TaskHandle;
-import com.google.appengine.api.taskqueue.TaskOptions;
-import com.google.appengine.api.taskqueue.TaskOptions.Method;
 import com.google.common.base.Joiner;
 import com.google.common.base.Splitter;
+import com.google.common.collect.ImmutableMultimap;
 import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.Sets;
 import com.google.common.flogger.FluentLogger;
@@ -35,6 +31,7 @@ import google.registry.export.datastore.DatastoreAdmin;
 import google.registry.export.datastore.Operation;
 import google.registry.model.annotations.DeleteAfterMigration;
 import google.registry.request.Action;
+import google.registry.request.Action.Service;
 import google.registry.request.HttpException;
 import google.registry.request.HttpException.BadRequestException;
 import google.registry.request.HttpException.InternalServerErrorException;
@@ -45,6 +42,7 @@ import google.registry.request.RequestMethod;
 import google.registry.request.Response;
 import google.registry.request.auth.Auth;
 import google.registry.util.Clock;
+import google.registry.util.CloudTasksUtils;
 import java.io.IOException;
 import java.util.Set;
 import javax.inject.Inject;
@@ -83,6 +81,7 @@ public class CheckBackupAction implements Runnable {
   @Inject DatastoreAdmin datastoreAdmin;
   @Inject Clock clock;
   @Inject Response response;
+  @Inject CloudTasksUtils cloudTasksUtils;
   @Inject @RequestMethod Action.Method requestMethod;
 
   @Inject
@@ -175,21 +174,22 @@ public class CheckBackupAction implements Runnable {
     if (exportedKindsToLoad.isEmpty()) {
       message += "no kinds to load into BigQuery.";
     } else {
-      enqueueUploadBackupTask(backupId, backup.getExportFolderUrl(), exportedKindsToLoad);
+      /** Enqueue a task for starting a backup load. */
+      cloudTasksUtils.enqueue(
+          UploadDatastoreBackupAction.QUEUE,
+          cloudTasksUtils.createPostTask(
+              UploadDatastoreBackupAction.PATH,
+              Service.BACKEND.toString(),
+              ImmutableMultimap.of(
+                  UploadDatastoreBackupAction.UPLOAD_BACKUP_ID_PARAM,
+                  backupId,
+                  UploadDatastoreBackupAction.UPLOAD_BACKUP_FOLDER_PARAM,
+                  backup.getExportFolderUrl(),
+                  UploadDatastoreBackupAction.UPLOAD_BACKUP_KINDS_PARAM,
+                  Joiner.on(',').join(exportedKindsToLoad))));
       message += "BigQuery load task enqueued.";
     }
     logger.atInfo().log(message);
     response.setPayload(message);
   }
-
-  /** Enqueue a poll task to monitor the named backup for completion. */
-  static TaskHandle enqueuePollTask(String backupId, ImmutableSet<String> kindsToLoad) {
-    return QueueFactory.getQueue(QUEUE)
-        .add(
-            TaskOptions.Builder.withUrl(PATH)
-                .method(Method.POST)
-                .countdownMillis(POLL_COUNTDOWN.getMillis())
-                .param(CHECK_BACKUP_NAME_PARAM, backupId)
-                .param(CHECK_BACKUP_KINDS_TO_LOAD_PARAM, Joiner.on(',').join(kindsToLoad)));
-  }
 }

core/src/main/java/google/registry/export/UpdateSnapshotViewAction.java

@@ -21,8 +21,6 @@ import com.google.api.services.bigquery.Bigquery;
 import com.google.api.services.bigquery.model.Table;
 import com.google.api.services.bigquery.model.TableReference;
 import com.google.api.services.bigquery.model.ViewDefinition;
-import com.google.appengine.api.taskqueue.TaskOptions;
-import com.google.appengine.api.taskqueue.TaskOptions.Method;
 import com.google.common.flogger.FluentLogger;
 import google.registry.bigquery.CheckedBigquery;
 import google.registry.config.RegistryConfig.Config;
@@ -85,17 +83,6 @@ public class UpdateSnapshotViewAction implements Runnable {
   @Inject
   UpdateSnapshotViewAction() {}
 
-  /** Create a task for updating a snapshot view. */
-  static TaskOptions createViewUpdateTask(
-      String datasetId, String tableId, String kindName, String viewName) {
-    return TaskOptions.Builder.withUrl(PATH)
-        .method(Method.POST)
-        .param(UPDATE_SNAPSHOT_DATASET_ID_PARAM, datasetId)
-        .param(UPDATE_SNAPSHOT_TABLE_ID_PARAM, tableId)
-        .param(UPDATE_SNAPSHOT_KIND_PARAM, kindName)
-        .param(UPDATE_SNAPSHOT_VIEWNAME_PARAM, viewName);
-  }
-
   @Override
   public void run() {
     try {

core/src/main/java/google/registry/export/UploadDatastoreBackupAction.java

@@ -14,9 +14,7 @@
 
 package google.registry.export;
 
-import static com.google.appengine.api.taskqueue.QueueFactory.getQueue;
 import static com.google.common.base.MoreObjects.firstNonNull;
-import static google.registry.export.UpdateSnapshotViewAction.createViewUpdateTask;
 import static google.registry.request.Action.Method.POST;
 
 import com.google.api.services.bigquery.Bigquery;
@@ -25,27 +23,33 @@ import com.google.api.services.bigquery.model.JobConfiguration;
 import com.google.api.services.bigquery.model.JobConfigurationLoad;
 import com.google.api.services.bigquery.model.JobReference;
 import com.google.api.services.bigquery.model.TableReference;
-import com.google.appengine.api.taskqueue.TaskHandle;
-import com.google.appengine.api.taskqueue.TaskOptions;
-import com.google.appengine.api.taskqueue.TaskOptions.Method;
+import com.google.cloud.tasks.v2.Task;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Joiner;
 import com.google.common.base.Splitter;
 import com.google.common.collect.ImmutableList;
-import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.ImmutableMultimap;
 import com.google.common.flogger.FluentLogger;
+import com.google.common.net.HttpHeaders;
+import com.google.common.net.MediaType;
+import com.google.protobuf.ByteString;
+import com.google.protobuf.util.Timestamps;
 import google.registry.bigquery.BigqueryUtils.SourceFormat;
 import google.registry.bigquery.BigqueryUtils.WriteDisposition;
 import google.registry.bigquery.CheckedBigquery;
 import google.registry.config.RegistryConfig.Config;
-import google.registry.export.BigqueryPollJobAction.BigqueryPollJobEnqueuer;
 import google.registry.model.annotations.DeleteAfterMigration;
 import google.registry.request.Action;
+import google.registry.request.Action.Service;
 import google.registry.request.HttpException.BadRequestException;
 import google.registry.request.HttpException.InternalServerErrorException;
 import google.registry.request.Parameter;
 import google.registry.request.auth.Auth;
+import google.registry.util.Clock;
+import google.registry.util.CloudTasksUtils;
+import java.io.ByteArrayOutputStream;
 import java.io.IOException;
+import java.io.ObjectOutputStream;
 import javax.inject.Inject;
 
 /** Action to load a Datastore backup from Google Cloud Storage into BigQuery. */
@@ -75,7 +79,9 @@ public class UploadDatastoreBackupAction implements Runnable {
   private static final FluentLogger logger = FluentLogger.forEnclosingClass();
 
   @Inject CheckedBigquery checkedBigquery;
-  @Inject BigqueryPollJobEnqueuer bigqueryPollEnqueuer;
+  @Inject CloudTasksUtils cloudTasksUtils;
+  @Inject Clock clock;
+
   @Inject @Config("projectId") String projectId;
 
   @Inject
@@ -93,18 +99,6 @@ public class UploadDatastoreBackupAction implements Runnable {
   @Inject
   UploadDatastoreBackupAction() {}
 
-  /** Enqueue a task for starting a backup load. */
-  public static TaskHandle enqueueUploadBackupTask(
-      String backupId, String gcsFile, ImmutableSet<String> kinds) {
-    return getQueue(QUEUE)
-        .add(
-            TaskOptions.Builder.withUrl(PATH)
-                .method(Method.POST)
-                .param(UPLOAD_BACKUP_ID_PARAM, backupId)
-                .param(UPLOAD_BACKUP_FOLDER_PARAM, gcsFile)
-                .param(UPLOAD_BACKUP_KINDS_PARAM, Joiner.on(',').join(kinds)));
-  }
-
   @Override
   public void run() {
     try {
@@ -142,12 +136,46 @@ public class UploadDatastoreBackupAction implements Runnable {
       Job job = makeLoadJob(jobRef, sourceUri, tableId);
       bigquery.jobs().insert(projectId, job).execute();
 
-      // Enqueue a task to check on the load job's completion, and if it succeeds, to update a
-      // well-known view in BigQuery to point at the newly loaded backup table for this kind.
-      bigqueryPollEnqueuer.enqueuePollTask(
-          jobRef,
-          createViewUpdateTask(BACKUP_DATASET, tableId, kindName, LATEST_BACKUP_VIEW_NAME),
-          getQueue(UpdateSnapshotViewAction.QUEUE));
+      // Serialize the chainedTask into a byte array to put in the task payload.
+      ByteArrayOutputStream taskBytes = new ByteArrayOutputStream();
+      new ObjectOutputStream(taskBytes)
+          .writeObject(
+              cloudTasksUtils.createPostTask(
+                  UpdateSnapshotViewAction.PATH,
+                  Service.BACKEND.toString(),
+                  ImmutableMultimap.of(
+                      UpdateSnapshotViewAction.UPDATE_SNAPSHOT_DATASET_ID_PARAM,
+                      BACKUP_DATASET,
+                      UpdateSnapshotViewAction.UPDATE_SNAPSHOT_TABLE_ID_PARAM,
+                      tableId,
+                      UpdateSnapshotViewAction.UPDATE_SNAPSHOT_KIND_PARAM,
+                      kindName,
+                      UpdateSnapshotViewAction.UPDATE_SNAPSHOT_VIEWNAME_PARAM,
+                      LATEST_BACKUP_VIEW_NAME)));
+
+      // Enqueues a task to poll for the success or failure of the referenced BigQuery job and to
+      // launch the provided task in the specified queue if the job succeeds.
+      cloudTasksUtils.enqueue(
+          BigqueryPollJobAction.QUEUE,
+          Task.newBuilder()
+              .setAppEngineHttpRequest(
+                  cloudTasksUtils
+                      .createPostTask(BigqueryPollJobAction.PATH, Service.BACKEND.toString(), null)
+                      .getAppEngineHttpRequest()
+                      .toBuilder()
+                      .putHeaders(BigqueryPollJobAction.PROJECT_ID_HEADER, jobRef.getProjectId())
+                      .putHeaders(BigqueryPollJobAction.JOB_ID_HEADER, jobRef.getJobId())
+                      .putHeaders(
+                          BigqueryPollJobAction.CHAINED_TASK_QUEUE_HEADER,
+                          UpdateSnapshotViewAction.QUEUE)
+                      // need to include CONTENT_TYPE in header when body is not empty
+                      .putHeaders(HttpHeaders.CONTENT_TYPE, MediaType.FORM_DATA.toString())
+                      .setBody(ByteString.copyFrom(taskBytes.toByteArray()))
+                      .build())
+              .setScheduleTime(
+                  Timestamps.fromMillis(
+                      clock.nowUtc().plus(BigqueryPollJobAction.POLL_COUNTDOWN).getMillis()))
+              .build());
 
       builder.append(String.format(" - %s:%s\n", projectId, jobId));
       logger.atInfo().log("Submitted load job %s:%s.", projectId, jobId);

core/src/main/java/google/registry/model/EntityClasses.java

@@ -40,6 +40,7 @@ import google.registry.model.rde.RdeRevision;
 import google.registry.model.registrar.Registrar;
 import google.registry.model.registrar.RegistrarContact;
 import google.registry.model.replay.LastSqlTransaction;
+import google.registry.model.replay.ReplayGap;
 import google.registry.model.reporting.HistoryEntry;
 import google.registry.model.server.Lock;
 import google.registry.model.server.ServerSecret;
@@ -86,6 +87,7 @@ public final class EntityClasses {
           Registrar.class,
           RegistrarContact.class,
           Registry.class,
+          ReplayGap.class,
           ServerSecret.class);
 
   private EntityClasses() {}

core/src/main/java/google/registry/model/billing/BillingEvent.java

@@ -308,7 +308,9 @@ public abstract class BillingEvent extends ImmutableObject
         @javax.persistence.Index(columnList = "eventTime"),
         @javax.persistence.Index(columnList = "billingTime"),
         @javax.persistence.Index(columnList = "syntheticCreationTime"),
-        @javax.persistence.Index(columnList = "allocationToken")
+        @javax.persistence.Index(columnList = "domainRepoId"),
+        @javax.persistence.Index(columnList = "allocationToken"),
+        @javax.persistence.Index(columnList = "cancellation_matching_billing_recurrence_id")
       })
   @AttributeOverride(name = "id", column = @Column(name = "billing_event_id"))
   @WithLongVKey(compositeKey = true)
@@ -518,6 +520,7 @@ public abstract class BillingEvent extends ImmutableObject
       indexes = {
         @javax.persistence.Index(columnList = "registrarId"),
         @javax.persistence.Index(columnList = "eventTime"),
+        @javax.persistence.Index(columnList = "domainRepoId"),
         @javax.persistence.Index(columnList = "recurrenceEndTime"),
         @javax.persistence.Index(columnList = "recurrence_time_of_year")
       })
@@ -614,7 +617,10 @@ public abstract class BillingEvent extends ImmutableObject
       indexes = {
         @javax.persistence.Index(columnList = "registrarId"),
         @javax.persistence.Index(columnList = "eventTime"),
-        @javax.persistence.Index(columnList = "billingTime")
+        @javax.persistence.Index(columnList = "domainRepoId"),
+        @javax.persistence.Index(columnList = "billingTime"),
+        @javax.persistence.Index(columnList = "billing_event_id"),
+        @javax.persistence.Index(columnList = "billing_recurrence_id")
       })
   @AttributeOverride(name = "id", column = @Column(name = "billing_cancellation_id"))
   @WithLongVKey(compositeKey = true)

core/src/main/java/google/registry/model/domain/DomainBase.java

@@ -69,7 +69,10 @@ import org.joda.time.DateTime;
       @Index(columnList = "techContact"),
       @Index(columnList = "tld"),
       @Index(columnList = "registrantContact"),
-      @Index(columnList = "dnsRefreshRequestTime")
+      @Index(columnList = "dnsRefreshRequestTime"),
+      @Index(columnList = "billing_recurrence_id"),
+      @Index(columnList = "transfer_billing_event_id"),
+      @Index(columnList = "transfer_billing_recurrence_id")
     })
 @WithStringVKey
 @ExternalMessagingName("domain")
@@ -89,7 +92,10 @@ public class DomainBase extends DomainContent
   @ElementCollection
   @JoinTable(
       name = "DomainHost",
-      indexes = {@Index(columnList = "domain_repo_id,host_repo_id", unique = true)})
+      indexes = {
+        @Index(columnList = "domain_repo_id,host_repo_id", unique = true),
+        @Index(columnList = "host_repo_id")
+      })
   @Access(AccessType.PROPERTY)
   @Column(name = "host_repo_id")
   public Set<VKey<HostResource>> getNsHosts() {

core/src/main/java/google/registry/model/domain/GracePeriod.java

@@ -46,7 +46,12 @@ import org.joda.time.DateTime;
  */
 @Embed
 @Entity
-@Table(indexes = @Index(columnList = "domainRepoId"))
+@Table(
+    indexes = {
+      @Index(columnList = "domainRepoId"),
+      @Index(columnList = "billing_event_id"),
+      @Index(columnList = "billing_recurrence_id")
+    })
 public class GracePeriod extends GracePeriodBase implements DatastoreAndSqlEntity {
 
   @Id

core/src/main/java/google/registry/model/domain/token/AllocationToken.java

@@ -76,6 +76,8 @@ import org.joda.time.DateTime;
       @javax.persistence.Index(
           columnList = "domainName",
           name = "allocation_token_domain_name_idx"),
+      @javax.persistence.Index(columnList = "tokenType"),
+      @javax.persistence.Index(columnList = "redemption_domain_repo_id")
     })
 public class AllocationToken extends BackupGroupRoot implements Buildable, DatastoreAndSqlEntity {
 

core/src/main/java/google/registry/model/host/HostResource.java

@@ -47,7 +47,12 @@ import javax.persistence.AccessType;
      * would prevent us from using JPA standard bootstrapping). For now, there is no obvious benefit
      * doing either.
      */
-    indexes = {@javax.persistence.Index(columnList = "hostName")})
+    indexes = {
+      @javax.persistence.Index(columnList = "hostName"),
+      @javax.persistence.Index(columnList = "creationTime"),
+      @javax.persistence.Index(columnList = "deletionTime"),
+      @javax.persistence.Index(columnList = "currentSponsorRegistrarId")
+    })
 @ExternalMessagingName("host")
 @WithStringVKey
 @Access(AccessType.FIELD) // otherwise it'll use the default if the repoId (property)

core/src/main/java/google/registry/model/replay/ReplayGap.java

@@ -0,0 +1,60 @@
+// Copyright 2022 The Nomulus Authors. All Rights Reserved.
+//
+// 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 google.registry.model.replay;
+
+import static google.registry.model.annotations.NotBackedUp.Reason.TRANSIENT;
+
+import com.googlecode.objectify.annotation.Entity;
+import com.googlecode.objectify.annotation.Id;
+import google.registry.model.ImmutableObject;
+import google.registry.model.annotations.DeleteAfterMigration;
+import google.registry.model.annotations.NotBackedUp;
+import org.joda.time.DateTime;
+
+/**
+ * Tracks gaps in transaction ids when replicating from SQL to datastore.
+ *
+ * <p>SQL -&gt; DS replication uses a Transaction table indexed by a SEQUENCE column, which normally
+ * increments monotonically for each committed transaction. Gaps in this sequence can occur when a
+ * transaction is rolled back or when a transaction has been initiated but not committed to the
+ * table at the time of a query. To protect us from the latter scenario, we need to keep track of
+ * these gaps and replay any of them that have been filled in since we processed their batch.
+ */
+@DeleteAfterMigration
+@NotBackedUp(reason = TRANSIENT)
+@Entity
+public class ReplayGap extends ImmutableObject implements DatastoreOnlyEntity {
+  @Id long transactionId;
+
+  // We can't use a CreateAutoTimestamp here because this ends up getting persisted in an ofy
+  // transaction that happens in JPA mode, so we don't end up getting an active transaction manager
+  // when the timestamp needs to be set.
+  DateTime timestamp;
+
+  ReplayGap() {}
+
+  ReplayGap(DateTime timestamp, long transactionId) {
+    this.timestamp = timestamp;
+    this.transactionId = transactionId;
+  }
+
+  long getTransactionId() {
+    return transactionId;
+  }
+
+  DateTime getTimestamp() {
+    return timestamp;
+  }
+}

core/src/main/java/google/registry/model/replay/ReplicateToDatastoreAction.java

@@ -14,6 +14,8 @@
 
 package google.registry.model.replay;
 
+import static com.google.common.base.Preconditions.checkState;
+import static com.google.common.collect.ImmutableList.toImmutableList;
 import static google.registry.model.ofy.ObjectifyService.auditedOfy;
 import static google.registry.persistence.transaction.TransactionManagerFactory.jpaTm;
 import static google.registry.persistence.transaction.TransactionManagerFactory.ofyTm;
@@ -40,6 +42,7 @@ import google.registry.request.auth.Auth;
 import google.registry.util.Clock;
 import google.registry.util.RequestStatusChecker;
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Optional;
 import javax.inject.Inject;
@@ -70,6 +73,20 @@ public class ReplicateToDatastoreAction implements Runnable {
    */
   public static final int BATCH_SIZE = 200;
 
+  /**
+   * The longest time that we'll keep trying to resolve a gap in the Transaction table in
+   * milliseconds, after which, the gap record will be deleted.
+   */
+  public static final long MAX_GAP_RETENTION_MILLIS = 300000;
+
+  /**
+   * The maximum number of entitities to be mutated per transaction. For our purposes, the entities
+   * that we're keeping track of are each in their own entity group. Per datastore documentation, we
+   * should be allowed to update up to 25 of them. In practice, we get an error if we go beyond 24
+   * (possibly due to something in our own infrastructure).
+   */
+  private static final int MAX_ENTITIES_PER_TXN = 24;
+
   public static final Duration REPLICATE_TO_DATASTORE_LOCK_LEASE_LENGTH = standardHours(1);
 
   private final Clock clock;
@@ -89,6 +106,12 @@ public class ReplicateToDatastoreAction implements Runnable {
     return getTransactionBatchAtSnapshot(Optional.empty());
   }
 
+  /**
+   * Get the next batch of transactions, optionally from a specific SQL database snapshot.
+   *
+   * <p>Note that this method may also apply transactions from previous batches that had not yet
+   * been committed at the time the previous batch was retrieved.
+   */
   static List<TransactionEntity> getTransactionBatchAtSnapshot(Optional<String> snapshotId) {
     // Get the next batch of transactions that we haven't replicated.
     LastSqlTransaction lastSqlTxnBeforeBatch = ofyTm().transact(LastSqlTransaction::load);
@@ -97,6 +120,11 @@ public class ReplicateToDatastoreAction implements Runnable {
           .transactWithoutBackup(
               () -> {
                 snapshotId.ifPresent(jpaTm()::setDatabaseSnapshot);
+
+                // Fill in any gaps in the transaction log that have since become available before
+                // processing the next batch.
+                applyMissingTransactions();
+
                 return jpaTm()
                     .query(
                         "SELECT txn FROM TransactionEntity txn WHERE id >" + " :lastId ORDER BY id",
@@ -111,61 +139,175 @@ public class ReplicateToDatastoreAction implements Runnable {
   }
 
   /**
-   * Apply a transaction to Datastore, returns true if there was a fatal error and the batch should
-   * be aborted.
+   * Iterate over the recent gaps in the Transaction table and apply any that have been filled in.
    *
-   * <p>Throws an exception if a fatal error occurred and the batch should be aborted
+   * <p>Must be called from within a JPA transaction.
+   *
+   * <p>Gap rewriting is a complicated matter, and the algorithm is the product of some very deep
+   * consideration by mmuller and weiminyu. Basically, the constraints are:
+   *
+   * <ol>
+   *   <li>Replay has to work against a database snapshot (gap replay would break this, so we don't
+   *       call this method when replaying against a snapshot)
+   * </ol>
+   */
+  private static void applyMissingTransactions() {
+    long now = jpaTm().getTransactionTime().getMillis();
+    ImmutableList<ReplayGap> gaps = ofyTm().loadAllOf(ReplayGap.class);
+    jpaTm()
+        .query("SELECT txn from TransactionEntity txn WHERE id IN :gapIds", TransactionEntity.class)
+        .setParameter(
+            "gapIds", gaps.stream().map(gap -> gap.getTransactionId()).collect(toImmutableList()))
+        .getResultStream()
+        .forEach(
+            txn -> {
+              // Transcribe the transaction and delete the gap record in the same ofy transaction.
+              ofyTm()
+                  .transact(
+                      () -> {
+                        // Write the transaction to datastore.
+                        try {
+                          Transaction.deserialize(txn.getContents()).writeToDatastore();
+                        } catch (IOException e) {
+                          throw new RuntimeException("Error during transaction deserialization", e);
+                        }
+
+                        // Find and delete the gap record.
+                        ImmutableList<ReplayGap> filledGaps =
+                            gaps.stream()
+                                .filter(gap -> gap.getTransactionId() == txn.getId())
+                                .collect(toImmutableList());
+                        checkState(
+                            filledGaps.size() == 1,
+                            "Bad list of gaps for discovered id: %s",
+                            filledGaps);
+                        auditedOfy().deleteIgnoringReadOnlyWithoutBackup().entity(gaps.get(0));
+                      });
+              logger.atInfo().log("Applied missing transaction %s", txn.getId());
+            });
+
+    // Clean up any gaps that have expired (in batches because they're each in their own entity
+    // group).
+    ArrayList<ReplayGap> gapBatch = new ArrayList<>();
+    gaps.stream()
+        .forEach(
+            gap -> {
+              if (now - gap.getTimestamp().getMillis() > MAX_GAP_RETENTION_MILLIS) {
+                gapBatch.add(gap);
+              }
+              if (gapBatch.size() == MAX_ENTITIES_PER_TXN) {
+                deleteReplayGaps(gapBatch);
+                gapBatch.clear();
+              }
+            });
+    if (!gapBatch.isEmpty()) {
+      deleteReplayGaps(gapBatch);
+    }
+  }
+
+  private static void deleteReplayGaps(ArrayList<ReplayGap> gapsToDelete) {
+    logger.atInfo().log(
+        "deleting gap records for %s",
+        gapsToDelete.stream().map(g -> g.getTransactionId()).collect(toImmutableList()));
+    ofyTm()
+        .transact(() -> auditedOfy().deleteIgnoringReadOnlyWithoutBackup().entities(gapsToDelete));
+  }
+
+  /**
+   * Apply a transaction to Datastore.
+   *
+   * <p>Throws an exception if a fatal error occurred and the batch should be aborted.
    */
   @VisibleForTesting
   public static void applyTransaction(TransactionEntity txnEntity) {
     logger.atInfo().log("Applying a single transaction Cloud SQL -> Cloud Datastore.");
+    boolean done = false;
     try (UpdateAutoTimestamp.DisableAutoUpdateResource disabler =
         UpdateAutoTimestamp.disableAutoUpdate()) {
-      ofyTm()
-          .transact(
-              () -> {
-                // Reload the last transaction id, which could possibly have changed.
-                LastSqlTransaction lastSqlTxn = LastSqlTransaction.load();
-                long nextTxnId = lastSqlTxn.getTransactionId() + 1;
 
-                // Skip missing transactions.  Missed transactions can happen normally.  If a
-                // transaction gets rolled back, the sequence counter doesn't.
-                while (nextTxnId < txnEntity.getId()) {
-                  logger.atWarning().log(
-                      "Ignoring transaction %s, which does not exist.", nextTxnId);
-                  ++nextTxnId;
-                }
+      // We put this in a do/while loop because we can potentially clean out some range of gaps
+      // first and we want to do those in their own transaction so as not to run up the entity group
+      // count.  (This is a highly pathological case: consecutive gaps are rare, but the fact that
+      // they can occur potentially increases the entity group count to beyond what we can
+      // accommodate in a single transaction.)
+      do {
+        done =
+            ofyTm()
+                .transact(
+                    () -> {
+                      // Reload the last transaction id, which could possibly have changed.
+                      LastSqlTransaction lastSqlTxn = LastSqlTransaction.load();
+                      long nextTxnId = lastSqlTxn.getTransactionId() + 1;
 
-                if (nextTxnId > txnEntity.getId()) {
-                  // We've already replayed this transaction.  This shouldn't happen, as GAE cron
-                  // is supposed to avoid overruns and this action shouldn't be executed from any
-                  // other context, but it's not harmful as we can just ignore the transaction.  Log
-                  // it so that we know about it and move on.
-                  logger.atWarning().log(
-                      "Ignoring transaction %s, which appears to have already been applied.",
-                      txnEntity.getId());
-                  return;
-                }
+                      // Skip missing transactions.  Missed transactions can happen normally.  If a
+                      // transaction gets rolled back, the sequence counter doesn't.
+                      int gapCount = 0;
+                      while (nextTxnId < txnEntity.getId()) {
+                        logger.atWarning().log(
+                            "Ignoring transaction %s, which does not exist.", nextTxnId);
+                        auditedOfy()
+                            .saveIgnoringReadOnlyWithoutBackup()
+                            .entity(new ReplayGap(ofyTm().getTransactionTime(), nextTxnId));
+                        ++nextTxnId;
 
-                logger.atInfo().log(
-                    "Applying transaction %s to Cloud Datastore.", txnEntity.getId());
+                        // Don't exceed the entity group count trying to clean these up (we stop at
+                        // max
+                        // - 1 because we also want to save the lastSqlTransaction).
+                        if (++gapCount == MAX_ENTITIES_PER_TXN - 1) {
+                          break;
+                        }
+                      }
 
-                // At this point, we know txnEntity is the correct next transaction, so write it
-                // to Datastore.
-                try {
-                  Transaction.deserialize(txnEntity.getContents()).writeToDatastore();
-                } catch (IOException e) {
-                  throw new RuntimeException("Error during transaction deserialization", e);
-                }
+                      // Don't write gap records in the same transaction as the SQL transaction that
+                      // we're replaying.  Return false to force us to go through and repeat in a
+                      // new
+                      // transaction.
+                      if (gapCount > 0) {
+                        // We haven't replayed the transaction, but we've determined that we can
+                        // ignore everything before it so update lastSqlTransaction accordingly.
+                        auditedOfy()
+                            .saveIgnoringReadOnlyWithoutBackup()
+                            .entity(lastSqlTxn.cloneWithNewTransactionId(nextTxnId - 1));
+                        return false;
+                      }
 
-                // Write the updated last transaction id to Datastore as part of this Datastore
-                // transaction.
-                auditedOfy()
-                    .saveIgnoringReadOnlyWithoutBackup()
-                    .entity(lastSqlTxn.cloneWithNewTransactionId(nextTxnId));
-                logger.atInfo().log(
-                    "Finished applying single transaction Cloud SQL -> Cloud Datastore.");
-              });
+                      if (nextTxnId > txnEntity.getId()) {
+                        // We've already replayed this transaction.  This shouldn't happen, as GAE
+                        // cron
+                        // is supposed to avoid overruns and this action shouldn't be executed from
+                        // any
+                        // other context, but it's not harmful as we can just ignore the
+                        // transaction.
+                        // Log it so that we know about it and move on.
+                        logger.atWarning().log(
+                            "Ignoring transaction %s, which appears to have already been applied.",
+                            txnEntity.getId());
+                        return true;
+                      }
+
+                      logger.atInfo().log(
+                          "Applying transaction %s to Cloud Datastore.", txnEntity.getId());
+
+                      // At this point, we know txnEntity is the correct next transaction, so write
+                      // it
+                      // to Datastore.
+                      try {
+                        Transaction.deserialize(txnEntity.getContents()).writeToDatastore();
+                      } catch (IOException e) {
+                        throw new RuntimeException("Error during transaction deserialization", e);
+                      }
+
+                      // Write the updated last transaction id to Datastore as part of this
+                      // Datastore
+                      // transaction.
+                      auditedOfy()
+                          .saveIgnoringReadOnlyWithoutBackup()
+                          .entity(lastSqlTxn.cloneWithNewTransactionId(nextTxnId));
+                      logger.atInfo().log(
+                          "Finished applying single transaction Cloud SQL -> Cloud Datastore.");
+                      return true;
+                    });
+      } while (!done);
     }
   }
 

core/src/main/java/google/registry/model/tld/Registries.java

@@ -24,6 +24,7 @@ import static com.google.common.collect.Maps.filterValues;
 import static google.registry.model.CacheUtils.memoizeWithShortExpiration;
 import static google.registry.model.common.EntityGroupRoot.getCrossTldKey;
 import static google.registry.model.ofy.ObjectifyService.auditedOfy;
+import static google.registry.persistence.transaction.TransactionManagerFactory.jpaTm;
 import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
 import static google.registry.util.CollectionUtils.entriesToImmutableMap;
 import static google.registry.util.PreconditionsUtils.checkArgumentNotNull;
@@ -38,6 +39,8 @@ import com.google.common.net.InternetDomainName;
 import com.googlecode.objectify.Key;
 import google.registry.model.tld.Registry.TldType;
 import java.util.Optional;
+import java.util.stream.Stream;
+import javax.persistence.EntityManager;
 
 /** Utilities for finding and listing {@link Registry} entities. */
 public final class Registries {
@@ -58,23 +61,31 @@ public final class Registries {
         () ->
             tm().doTransactionless(
                     () -> {
-                      ImmutableSet<String> tlds =
-                          tm().isOfy()
-                              ? auditedOfy()
-                                  .load()
-                                  .type(Registry.class)
-                                  .ancestor(getCrossTldKey())
-                                  .keys()
-                                  .list()
-                                  .stream()
-                                  .map(Key::getName)
-                                  .collect(toImmutableSet())
-                              : tm().loadAllOf(Registry.class).stream()
-                                  .map(Registry::getTldStr)
-                                  .collect(toImmutableSet());
-                      return Registry.getAll(tlds).stream()
-                          .map(e -> Maps.immutableEntry(e.getTldStr(), e.getTldType()))
-                          .collect(entriesToImmutableMap());
+                      if (tm().isOfy()) {
+                        ImmutableSet<String> tlds =
+                            auditedOfy()
+                                .load()
+                                .type(Registry.class)
+                                .ancestor(getCrossTldKey())
+                                .keys()
+                                .list()
+                                .stream()
+                                .map(Key::getName)
+                                .collect(toImmutableSet());
+                        return Registry.getAll(tlds).stream()
+                            .map(e -> Maps.immutableEntry(e.getTldStr(), e.getTldType()))
+                            .collect(entriesToImmutableMap());
+                      } else {
+                        EntityManager entityManager = jpaTm().getEntityManager();
+                        Stream<?> resultStream =
+                            entityManager
+                                .createQuery("SELECT tldStrId, tldType FROM Tld")
+                                .getResultStream();
+                        return resultStream
+                            .map(e -> ((Object[]) e))
+                            .map(e -> Maps.immutableEntry((String) e[0], ((TldType) e[1])))
+                            .collect(entriesToImmutableMap());
+                      }
                     }));
   }
 

core/src/main/java/google/registry/model/transfer/DomainTransferData.java

@@ -94,8 +94,6 @@ public class DomainTransferData extends TransferData<DomainTransferData.Builder>
    *
    * <p>This field should be null if there is not currently a pending transfer or if the object
    * being transferred is not a domain.
-   *
-   * <p>TODO(b/158230654) Remove unused columns for TransferData in Contact table.
    */
   @IgnoreSave(IfNull.class)
   @Column(name = "transfer_billing_event_id")

core/src/main/java/google/registry/rde/BrdaCopyAction.java

@@ -14,8 +14,13 @@
 
 package google.registry.rde;
 
+import static google.registry.model.common.Cursor.CursorType.BRDA;
+import static google.registry.model.common.Cursor.getCursorTimeOrStartOfTime;
 import static google.registry.model.rde.RdeMode.THIN;
+import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
+import static google.registry.persistence.transaction.TransactionManagerUtil.transactIfJpaTm;
 import static google.registry.request.Action.Method.POST;
+import static google.registry.util.DateTimeUtils.isBeforeOrAt;
 
 import com.google.cloud.storage.BlobId;
 import com.google.common.flogger.FluentLogger;
@@ -23,9 +28,11 @@ import com.google.common.io.ByteStreams;
 import google.registry.config.RegistryConfig.Config;
 import google.registry.gcs.GcsUtils;
 import google.registry.keyring.api.KeyModule.Key;
+import google.registry.model.common.Cursor;
 import google.registry.model.rde.RdeNamingUtils;
 import google.registry.model.rde.RdeRevision;
 import google.registry.request.Action;
+import google.registry.request.HttpException.NoContentException;
 import google.registry.request.Parameter;
 import google.registry.request.RequestParameters;
 import google.registry.request.auth.Auth;
@@ -89,6 +96,16 @@ public final class BrdaCopyAction implements Runnable {
   private void copyAsRyde() throws IOException {
     // TODO(b/217772483): consider guarding this action with a lock and check if there is work.
     // Not urgent since file writes on GCS are atomic.
+    Optional<Cursor> cursor =
+        transactIfJpaTm(() -> tm().loadByKeyIfPresent(Cursor.createVKey(BRDA, tld)));
+    DateTime brdaCursorTime = getCursorTimeOrStartOfTime(cursor);
+    if (isBeforeOrAt(brdaCursorTime, watermark)) {
+      throw new NoContentException(
+          String.format(
+              "Waiting on RdeStagingAction for TLD %s to copy BRDA deposit for %s to GCS; "
+                  + "last BRDA staging completion was before %s",
+              tld, watermark, brdaCursorTime));
+    }
     int revision =
         RdeRevision.getCurrentRevision(tld, watermark, THIN)
             .orElseThrow(

core/src/main/java/google/registry/rde/RdeStagingReducer.java

@@ -64,6 +64,7 @@ public final class RdeStagingReducer extends Reducer<PendingDeposit, DepositFrag
   private static final long serialVersionUID = 60326234579091203L;
 
   private static final FluentLogger logger = FluentLogger.forEnclosingClass();
+  private static final Duration ENQUEUE_DELAY = Duration.standardMinutes(1);
 
   private final CloudTasksUtils cloudTasksUtils;
   private final LockHandler lockHandler;
@@ -202,6 +203,14 @@ public final class RdeStagingReducer extends Reducer<PendingDeposit, DepositFrag
       logger.atInfo().log("Manual operation; not advancing cursor or enqueuing upload task.");
       return;
     }
+    // We need to save the revision in a separate transaction because the subsequent upload/copy
+    // action reads the most current revision from the database. If it is done in the same
+    // transaction with the enqueueing, the action might start running before the transaction is
+    // committed, due to Cloud Tasks not being transaction aware, unlike Task Queue. The downside
+    // is that if for some reason the second transaction is rolled back, the revision update is not
+    // undone. But this should be fine since the next run will just increment the revision and start
+    // over.
+    tm().transact(() -> RdeRevision.saveRevision(tld, watermark, mode, revision));
     tm().transact(
             () -> {
               Registry registry = Registry.get(tld);
@@ -225,29 +234,33 @@ public final class RdeStagingReducer extends Reducer<PendingDeposit, DepositFrag
               tm().put(Cursor.create(key.cursor(), newPosition, registry));
               logger.atInfo().log(
                   "Rolled forward %s on %s cursor to %s.", key.cursor(), tld, newPosition);
-              RdeRevision.saveRevision(tld, watermark, mode, revision);
               // Enqueueing a task is a side effect that is not undone if the transaction rolls
-              // back. So this may result in multiple copies of the same task being processed. This
-              // is fine because the RdeUploadAction is guarded by a lock and tracks progress by
-              // cursor. The BrdaCopyAction writes a file to GCS, which is an atomic action.
+              // back. So this may result in multiple copies of the same task being processed.
+              // This is fine because the RdeUploadAction is guarded by a lock and tracks progress
+              // by cursor. The BrdaCopyAction writes a file to GCS, which is an atomic action. It
+              // is also guarded by a cursor to not run before the cursor is updated. We also
+              // include a delay to minimize the chance that the enqueued job executes before the
+              // transaction is committed, which triggers a retry.
               if (mode == RdeMode.FULL) {
                 cloudTasksUtils.enqueue(
                     "rde-upload",
-                    cloudTasksUtils.createPostTask(
+                    cloudTasksUtils.createPostTaskWithDelay(
                         RdeUploadAction.PATH,
                         Service.BACKEND.toString(),
-                        ImmutableMultimap.of(RequestParameters.PARAM_TLD, tld)));
+                        ImmutableMultimap.of(RequestParameters.PARAM_TLD, tld),
+                        ENQUEUE_DELAY));
               } else {
                 cloudTasksUtils.enqueue(
                     "brda",
-                    cloudTasksUtils.createPostTask(
+                    cloudTasksUtils.createPostTaskWithDelay(
                         BrdaCopyAction.PATH,
                         Service.BACKEND.toString(),
                         ImmutableMultimap.of(
                             RequestParameters.PARAM_TLD,
                             tld,
                             RdeModule.PARAM_WATERMARK,
-                            watermark.toString())));
+                            watermark.toString()),
+                        ENQUEUE_DELAY));
               }
             });
   }

core/src/main/java/google/registry/rde/RdeUploadAction.java

@@ -148,7 +148,7 @@ public final class RdeUploadAction implements Runnable, EscrowTask {
       throw new NoContentException(
           String.format(
               "Waiting on RdeStagingAction for TLD %s to send %s upload; "
-                  + "last RDE staging completion was at %s",
+                  + "last RDE staging completion was before %s",
               tld, watermark, stagingCursorTime));
     }
     DateTime sftpCursorTime =

core/src/main/java/google/registry/request/RequestModule.java

@@ -30,6 +30,7 @@ import com.google.common.collect.ImmutableSet;
 import com.google.common.io.ByteStreams;
 import com.google.common.io.CharStreams;
 import com.google.common.net.MediaType;
+import com.google.protobuf.ByteString;
 import dagger.Module;
 import dagger.Provides;
 import google.registry.model.common.DatabaseMigrationStateSchedule.PrimaryDatabase;
@@ -184,6 +185,16 @@ public final class RequestModule {
     }
   }
 
+  @Provides
+  @Payload
+  static ByteString providePayloadAsByteString(HttpServletRequest req) {
+    try {
+      return ByteString.copyFrom(ByteStreams.toByteArray(req.getInputStream()));
+    } catch (IOException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
   @Provides
   static LockHandler provideLockHandler(LockHandlerImpl lockHandler) {
     return lockHandler;

core/src/main/java/google/registry/tools/DigestType.java

@@ -14,6 +14,9 @@
 
 package google.registry.tools;
 
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Maps;
+import java.util.Arrays;
 import java.util.Optional;
 
 /**
@@ -43,14 +46,12 @@ public enum DigestType {
     this.bytes = bytes;
   }
 
+  private static final ImmutableMap<Integer, DigestType> WIRE_VALUE_TO_DIGEST_TYPE =
+      Maps.uniqueIndex(Arrays.stream(DigestType.values()).iterator(), DigestType::getWireValue);
+
   /** Fetches a DigestType enumeration constant by its IANA assigned value. */
   public static Optional<DigestType> fromWireValue(int wireValue) {
-    for (DigestType alg : DigestType.values()) {
-      if (alg.getWireValue() == wireValue) {
-        return Optional.of(alg);
-      }
-    }
-    return Optional.empty();
+    return Optional.ofNullable(WIRE_VALUE_TO_DIGEST_TYPE.get(wireValue));
   }
 
   /** Fetches a value in the range [0, 255] that encodes this DS digest type on the wire. */

core/src/main/java/google/registry/tools/GenerateEscrowDepositCommand.java

@@ -33,7 +33,6 @@ import google.registry.model.rde.RdeMode;
 import google.registry.rde.RdeStagingAction;
 import google.registry.request.Action.Service;
 import google.registry.tools.params.DateTimeParameter;
-import google.registry.util.AppEngineServiceUtils;
 import google.registry.util.CloudTasksUtils;
 import java.util.List;
 import java.util.stream.Collectors;
@@ -90,8 +89,6 @@ final class GenerateEscrowDepositCommand implements CommandWithRemoteApi {
       required = true)
   private String outdir;
 
-  @Inject AppEngineServiceUtils appEngineServiceUtils;
-
   @Inject CloudTasksUtils cloudTasksUtils;
 
   @Override

core/src/main/java/google/registry/ui/server/registrar/RegistrarSettingsAction.java

@@ -59,7 +59,6 @@ import google.registry.ui.forms.FormException;
 import google.registry.ui.forms.FormFieldException;
 import google.registry.ui.server.RegistrarFormFields;
 import google.registry.ui.server.SendEmailUtils;
-import google.registry.util.AppEngineServiceUtils;
 import google.registry.util.CloudTasksUtils;
 import google.registry.util.CollectionUtils;
 import google.registry.util.DiffUtils;
@@ -108,7 +107,6 @@ public class RegistrarSettingsAction implements Runnable, JsonActionRunner.JsonA
   private static ThreadLocal<Boolean> isInTestDriver = ThreadLocal.withInitial(() -> false);
 
   @Inject JsonActionRunner jsonActionRunner;
-  @Inject AppEngineServiceUtils appEngineServiceUtils;
   @Inject RegistrarConsoleMetrics registrarConsoleMetrics;
   @Inject SendEmailUtils sendEmailUtils;
   @Inject AuthenticatedRegistrarAccessor registrarAccessor;

core/src/test/java/google/registry/backup/CommitLogCheckpointActionTest.java

@@ -47,11 +47,10 @@ public class CommitLogCheckpointActionTest {
 
   private DateTime now = DateTime.now(UTC);
   private CommitLogCheckpointAction task = new CommitLogCheckpointAction();
-  private final CloudTasksHelper cloudTasksHelper = new CloudTasksHelper();
+  private final CloudTasksHelper cloudTasksHelper = new CloudTasksHelper(new FakeClock(now));
 
   @BeforeEach
   void beforeEach() {
-    task.clock = new FakeClock(now);
     task.strategy = strategy;
     task.cloudTasksUtils = cloudTasksHelper.getTestCloudTasksUtils();
     when(strategy.computeCheckpoint())
@@ -68,7 +67,8 @@ public class CommitLogCheckpointActionTest {
         new TaskMatcher()
             .url(ExportCommitLogDiffAction.PATH)
             .param(ExportCommitLogDiffAction.LOWER_CHECKPOINT_TIME_PARAM, START_OF_TIME.toString())
-            .param(ExportCommitLogDiffAction.UPPER_CHECKPOINT_TIME_PARAM, now.toString()));
+            .param(ExportCommitLogDiffAction.UPPER_CHECKPOINT_TIME_PARAM, now.toString())
+            .scheduleTime(now.plus(CommitLogCheckpointAction.ENQUEUE_DELAY_SECONDS)));
     assertThat(loadRoot().getLastWrittenTime()).isEqualTo(now);
   }
 
@@ -82,7 +82,8 @@ public class CommitLogCheckpointActionTest {
         new TaskMatcher()
             .url(ExportCommitLogDiffAction.PATH)
             .param(ExportCommitLogDiffAction.LOWER_CHECKPOINT_TIME_PARAM, oneMinuteAgo.toString())
-            .param(ExportCommitLogDiffAction.UPPER_CHECKPOINT_TIME_PARAM, now.toString()));
+            .param(ExportCommitLogDiffAction.UPPER_CHECKPOINT_TIME_PARAM, now.toString())
+            .scheduleTime(now.plus(CommitLogCheckpointAction.ENQUEUE_DELAY_SECONDS)));
     assertThat(loadRoot().getLastWrittenTime()).isEqualTo(now);
   }
 

core/src/test/java/google/registry/batch/RelockDomainActionTest.java

@@ -31,8 +31,6 @@ import static google.registry.testing.SqlHelper.saveRegistryLock;
 import static google.registry.tools.LockOrUnlockDomainCommand.REGISTRY_LOCK_STATUSES;
 import static javax.servlet.http.HttpServletResponse.SC_NO_CONTENT;
 import static javax.servlet.http.HttpServletResponse.SC_OK;
-import static org.mockito.Mockito.lenient;
-import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.verifyNoMoreInteractions;
 
@@ -51,7 +49,6 @@ import google.registry.testing.FakeResponse;
 import google.registry.testing.TestOfyAndSql;
 import google.registry.testing.UserInfo;
 import google.registry.tools.DomainLockUtils;
-import google.registry.util.AppEngineServiceUtils;
 import google.registry.util.EmailMessage;
 import google.registry.util.SendEmailService;
 import google.registry.util.StringGenerator.Alphabets;
@@ -111,11 +108,6 @@ public class RelockDomainActionTest {
             DOMAIN_NAME, CLIENT_ID, false, Optional.empty());
     assertThat(loadByEntity(domain).getStatusValues()).containsNoneIn(REGISTRY_LOCK_STATUSES);
 
-    AppEngineServiceUtils appEngineServiceUtils = mock(AppEngineServiceUtils.class);
-    lenient()
-        .when(appEngineServiceUtils.getServiceHostname("backend"))
-        .thenReturn("backend.hostname.fake");
-
     action = createAction(oldLock.getRevisionId());
   }
 

core/src/test/java/google/registry/batch/ResaveEntityActionTest.java

@@ -44,7 +44,6 @@ import google.registry.testing.DualDatabaseTest;
 import google.registry.testing.FakeClock;
 import google.registry.testing.InjectExtension;
 import google.registry.testing.TestOfyAndSql;
-import google.registry.util.AppEngineServiceUtils;
 import org.joda.time.DateTime;
 import org.joda.time.Duration;
 import org.junit.jupiter.api.BeforeEach;
@@ -66,7 +65,6 @@ public class ResaveEntityActionTest {
 
   @RegisterExtension public final InjectExtension inject = new InjectExtension();
 
-  @Mock private AppEngineServiceUtils appEngineServiceUtils;
   @Mock private Response response;
   private final FakeClock clock = new FakeClock(DateTime.parse("2016-02-11T10:00:00Z"));
   private AsyncTaskEnqueuer asyncTaskEnqueuer;

core/src/test/java/google/registry/batch/SendExpiringCertificateNotificationEmailActionTest.java

@@ -59,49 +59,18 @@ class SendExpiringCertificateNotificationEmailActionTest {
   private static final String EXPIRATION_WARNING_EMAIL_BODY_TEXT =
       " Dear %1$s,\n"
           + "\n"
-          + "    We would like to inform you that your %2$s SSL certificate will expire at\n"
-          + "    %3$s. Please take note that using expired certificates will prevent\n"
-          + "    successful Registry login.\n"
+          + "We would like to inform you that your %2$s certificate will expire at %3$s."
           + "\n"
-          + "    Kindly update your production account certificate within the support\n"
-          + "    console using the following steps:\n"
+          + " Kind update your account using the following steps: "
           + "\n"
-          + "      1. Navigate to support.registry.google and login using your\n"
-          + "    %4$s@registry.google credentials.\n"
-          + "          * If this is your first time logging in, you will be prompted to\n"
-          + "          reset your password, so please keep your new password safe.\n"
-          + "          * If you are already logged in with some other Google account(s) but\n"
-          + "          not your %4$s@registry.google account, you need to click on\n"
-          + "          “Add Account” and login using your %4$s@registry.google credentials.\n"
-          + "      2. Select “Settings > Security” from the left navigation bar.\n"
-          + "      3. Click “Edit” on the top left corner.\n"
-          + "      4. Enter your full certificate string\n"
-          + "         (including lines -----BEGIN CERTIFICATE----- and\n"
-          + "         -----END CERTIFICATE-----) in the box.\n"
-          + "      5. Click “Save”. If there are validation issues with the form, you will\n"
-          + "         be prompted to fix them and click “Save” again.\n"
-          + "\n"
-          + "    A failover SSL certificate can also be added in order to prevent connection\n"
-          + "    issues once your main certificate expires. Connecting with either of the\n"
-          + "    certificates will work with our production EPP server.\n"
-          + "\n"
-          + "    Further information about our EPP connection requirements can be found in\n"
-          + "    section 9.2 in the updated Technical Guide in your Google Drive folder.\n"
-          + "\n"
-          + "    Note that account certificate changes take a few minutes to become\n"
-          + "    effective and that the existing connections will remain unaffected by\n"
-          + "    the change.\n"
-          + "\n"
-          + "    If you also would like to update your OT&E account certificate, please send\n"
-          + "    an email from your primary or technical contact to\n"
-          + "    registry-support@google.com and include the full certificate string\n"
-          + "    (including lines -----BEGIN CERTIFICATE----- and -----END CERTIFICATE-----).\n"
-          + "\n"
-          + "    Regards,\n"
-          + "    Google Registry\n";
+          + "  1. Navigate to support and login using your %4$s@registry.example credentials.\n"
+          + "  2. Click Settings -> Privacy on the top left corner.\n"
+          + "  3. Click Edit and enter certificate string."
+          + "  3. Click Save"
+          + "Regards,"
+          + "Example Registry";
 
-  private static final String EXPIRATION_WARNING_EMAIL_SUBJECT_TEXT =
-      "[Important] Expiring SSL certificate for Google " + "Registry EPP connection";
+  private static final String EXPIRATION_WARNING_EMAIL_SUBJECT_TEXT = "Expiration Warning Email";
 
   @RegisterExtension
   public final AppEngineExtension appEngine =
@@ -623,11 +592,11 @@ class SendExpiringCertificateNotificationEmailActionTest {
     assertThat(emailBody).contains(registrarName);
     assertThat(emailBody).contains(certificateType.getDisplayName());
     assertThat(emailBody).contains(certExpirationDateStr);
-    assertThat(emailBody).contains(registrarId + "@registry.google");
-    assertThat(emailBody).doesNotContain("%1$s@registry.google");
-    assertThat(emailBody).doesNotContain("%2$s@registry.google");
-    assertThat(emailBody).doesNotContain("%3$s@registry.google");
-    assertThat(emailBody).doesNotContain("%4$s@registry.google");
+    assertThat(emailBody).contains(registrarId + "@registry.example");
+    assertThat(emailBody).doesNotContain("%1$s");
+    assertThat(emailBody).doesNotContain("%2$s");
+    assertThat(emailBody).doesNotContain("%3$s");
+    assertThat(emailBody).doesNotContain("%4$s");
   }
 
   @TestOfyAndSql

core/src/test/java/google/registry/export/BackupDatastoreActionTest.java

@@ -17,16 +17,19 @@ package google.registry.export;
 import static com.google.common.truth.Truth.assertThat;
 import static google.registry.export.CheckBackupAction.CHECK_BACKUP_KINDS_TO_LOAD_PARAM;
 import static google.registry.export.CheckBackupAction.CHECK_BACKUP_NAME_PARAM;
-import static google.registry.testing.TaskQueueHelper.assertTasksEnqueued;
 import static org.mockito.Mockito.when;
 
+import com.google.cloud.tasks.v2.HttpMethod;
 import com.google.common.base.Joiner;
+import com.google.protobuf.util.Timestamps;
 import google.registry.export.datastore.DatastoreAdmin;
 import google.registry.export.datastore.DatastoreAdmin.Export;
 import google.registry.export.datastore.Operation;
 import google.registry.testing.AppEngineExtension;
+import google.registry.testing.CloudTasksHelper;
+import google.registry.testing.CloudTasksHelper.TaskMatcher;
+import google.registry.testing.FakeClock;
 import google.registry.testing.FakeResponse;
-import google.registry.testing.TaskQueueHelper.TaskMatcher;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
@@ -46,13 +49,15 @@ public class BackupDatastoreActionTest {
   @Mock private Operation backupOperation;
 
   private final FakeResponse response = new FakeResponse();
+  private CloudTasksHelper cloudTasksHelper = new CloudTasksHelper();
   private final BackupDatastoreAction action = new BackupDatastoreAction();
 
   @BeforeEach
   void beforeEach() throws Exception {
     action.datastoreAdmin = datastoreAdmin;
     action.response = response;
-
+    action.cloudTasksUtils = cloudTasksHelper.getTestCloudTasksUtils();
+    action.clock = new FakeClock();
     when(datastoreAdmin.export(
             "gs://registry-project-id-datastore-backups", AnnotatedEntities.getBackupKinds()))
         .thenReturn(exportRequest);
@@ -66,7 +71,7 @@ public class BackupDatastoreActionTest {
   @Test
   void testBackup_enqueuesPollTask() {
     action.run();
-    assertTasksEnqueued(
+    cloudTasksHelper.assertTasksEnqueued(
         CheckBackupAction.QUEUE,
         new TaskMatcher()
             .url(CheckBackupAction.PATH)
@@ -74,7 +79,10 @@ public class BackupDatastoreActionTest {
             .param(
                 CHECK_BACKUP_KINDS_TO_LOAD_PARAM,
                 Joiner.on(",").join(AnnotatedEntities.getReportingKinds()))
-            .method("POST"));
+            .method(HttpMethod.POST)
+            .scheduleTime(
+                Timestamps.fromMillis(
+                    action.clock.nowUtc().plus(CheckBackupAction.POLL_COUNTDOWN).getMillis())));
     assertThat(response.getPayload())
         .isEqualTo(
             "Datastore backup started with name: "

core/src/test/java/google/registry/export/BigqueryPollJobActionTest.java

@@ -14,11 +14,7 @@
 
 package google.registry.export;
 
-import static com.google.appengine.api.taskqueue.QueueFactory.getQueue;
-import static com.google.common.collect.Iterables.getOnlyElement;
 import static com.google.common.truth.Truth.assertThat;
-import static google.registry.testing.TaskQueueHelper.assertNoTasksEnqueued;
-import static google.registry.testing.TaskQueueHelper.assertTasksEnqueued;
 import static google.registry.testing.TestLogHandlerUtils.assertLogMessage;
 import static java.nio.charset.StandardCharsets.UTF_8;
 import static java.util.logging.Level.INFO;
@@ -30,27 +26,22 @@ import static org.mockito.Mockito.when;
 import com.google.api.services.bigquery.Bigquery;
 import com.google.api.services.bigquery.model.ErrorProto;
 import com.google.api.services.bigquery.model.Job;
-import com.google.api.services.bigquery.model.JobReference;
 import com.google.api.services.bigquery.model.JobStatus;
-import com.google.appengine.api.taskqueue.TaskOptions;
-import com.google.appengine.api.taskqueue.TaskOptions.Method;
-import com.google.appengine.api.taskqueue.dev.QueueStateInfo.TaskStateInfo;
-import google.registry.export.BigqueryPollJobAction.BigqueryPollJobEnqueuer;
+import com.google.cloud.tasks.v2.AppEngineHttpRequest;
+import com.google.cloud.tasks.v2.HttpMethod;
+import com.google.cloud.tasks.v2.Task;
+import com.google.common.net.HttpHeaders;
+import com.google.common.net.MediaType;
+import com.google.protobuf.ByteString;
 import google.registry.request.HttpException.BadRequestException;
 import google.registry.request.HttpException.NotModifiedException;
 import google.registry.testing.AppEngineExtension;
-import google.registry.testing.FakeClock;
-import google.registry.testing.FakeSleeper;
-import google.registry.testing.TaskQueueHelper;
-import google.registry.testing.TaskQueueHelper.TaskMatcher;
+import google.registry.testing.CloudTasksHelper;
+import google.registry.testing.CloudTasksHelper.TaskMatcher;
 import google.registry.util.CapturingLogHandler;
 import google.registry.util.JdkLoggerConfig;
-import google.registry.util.Retrier;
-import google.registry.util.TaskQueueUtils;
-import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
-import java.io.ObjectInputStream;
 import java.io.ObjectOutputStream;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -66,8 +57,6 @@ public class BigqueryPollJobActionTest {
   private static final String PROJECT_ID = "project_id";
   private static final String JOB_ID = "job_id";
   private static final String CHAINED_QUEUE_NAME = UpdateSnapshotViewAction.QUEUE;
-  private static final TaskQueueUtils TASK_QUEUE_UTILS =
-      new TaskQueueUtils(new Retrier(new FakeSleeper(new FakeClock()), 1));
 
   private final Bigquery bigquery = mock(Bigquery.class);
   private final Bigquery.Jobs bigqueryJobs = mock(Bigquery.Jobs.class);
@@ -75,53 +64,20 @@ public class BigqueryPollJobActionTest {
 
   private final CapturingLogHandler logHandler = new CapturingLogHandler();
   private BigqueryPollJobAction action = new BigqueryPollJobAction();
+  private CloudTasksHelper cloudTasksHelper = new CloudTasksHelper();
 
   @BeforeEach
   void beforeEach() throws Exception {
     action.bigquery = bigquery;
     when(bigquery.jobs()).thenReturn(bigqueryJobs);
     when(bigqueryJobs.get(PROJECT_ID, JOB_ID)).thenReturn(bigqueryJobsGet);
-    action.taskQueueUtils = TASK_QUEUE_UTILS;
+    action.cloudTasksUtils = cloudTasksHelper.getTestCloudTasksUtils();
     action.projectId = PROJECT_ID;
     action.jobId = JOB_ID;
     action.chainedQueueName = () -> CHAINED_QUEUE_NAME;
     JdkLoggerConfig.getConfig(BigqueryPollJobAction.class).addHandler(logHandler);
   }
 
-  private static TaskMatcher newPollJobTaskMatcher(String method) {
-    return new TaskMatcher()
-        .method(method)
-        .url(BigqueryPollJobAction.PATH)
-        .header(BigqueryPollJobAction.PROJECT_ID_HEADER, PROJECT_ID)
-        .header(BigqueryPollJobAction.JOB_ID_HEADER, JOB_ID);
-  }
-
-  @Test
-  void testSuccess_enqueuePollTask() {
-    new BigqueryPollJobEnqueuer(TASK_QUEUE_UTILS).enqueuePollTask(
-        new JobReference().setProjectId(PROJECT_ID).setJobId(JOB_ID));
-    assertTasksEnqueued(BigqueryPollJobAction.QUEUE, newPollJobTaskMatcher("GET"));
-  }
-
-  @Test
-  void testSuccess_enqueuePollTask_withChainedTask() throws Exception {
-    TaskOptions chainedTask = TaskOptions.Builder
-        .withUrl("/_dr/something")
-        .method(Method.POST)
-        .header("X-Testing", "foo")
-        .param("testing", "bar");
-    new BigqueryPollJobEnqueuer(TASK_QUEUE_UTILS).enqueuePollTask(
-        new JobReference().setProjectId(PROJECT_ID).setJobId(JOB_ID),
-        chainedTask,
-        getQueue(CHAINED_QUEUE_NAME));
-    assertTasksEnqueued(BigqueryPollJobAction.QUEUE, newPollJobTaskMatcher("POST"));
-    TaskStateInfo taskInfo = getOnlyElement(
-        TaskQueueHelper.getQueueInfo(BigqueryPollJobAction.QUEUE).getTaskInfo());
-    ByteArrayInputStream taskBodyBytes = new ByteArrayInputStream(taskInfo.getBodyAsBytes());
-    TaskOptions taskOptions = (TaskOptions) new ObjectInputStream(taskBodyBytes).readObject();
-    assertThat(taskOptions).isEqualTo(chainedTask);
-  }
-
   @Test
   void testSuccess_jobCompletedSuccessfully() throws Exception {
     when(bigqueryJobsGet.execute()).thenReturn(
@@ -136,15 +92,20 @@ public class BigqueryPollJobActionTest {
     when(bigqueryJobsGet.execute()).thenReturn(
         new Job().setStatus(new JobStatus().setState("DONE")));
 
-    TaskOptions chainedTask =
-        TaskOptions.Builder.withUrl("/_dr/something")
-            .method(Method.POST)
-            .header("X-Testing", "foo")
-            .param("testing", "bar")
-            .taskName("my_task_name");
+    Task chainedTask =
+        Task.newBuilder()
+            .setName("my_task_name")
+            .setAppEngineHttpRequest(
+                AppEngineHttpRequest.newBuilder()
+                    .setHttpMethod(HttpMethod.POST)
+                    .setRelativeUri("/_dr/something")
+                    .putHeaders("X-Test", "foo")
+                    .putHeaders(HttpHeaders.CONTENT_TYPE, MediaType.FORM_DATA.toString())
+                    .setBody(ByteString.copyFromUtf8("testing=bar")))
+            .build();
     ByteArrayOutputStream bytes = new ByteArrayOutputStream();
     new ObjectOutputStream(bytes).writeObject(chainedTask);
-    action.payload = bytes.toByteArray();
+    action.payload = ByteString.copyFrom(bytes.toByteArray());
 
     action.run();
     assertLogMessage(
@@ -153,14 +114,15 @@ public class BigqueryPollJobActionTest {
         logHandler,
         INFO,
         "Added chained task my_task_name for /_dr/something to queue " + CHAINED_QUEUE_NAME);
-    assertTasksEnqueued(
+    cloudTasksHelper.assertTasksEnqueued(
         CHAINED_QUEUE_NAME,
         new TaskMatcher()
             .url("/_dr/something")
-            .method("POST")
-            .header("X-Testing", "foo")
+            .header("X-Test", "foo")
+            .header(HttpHeaders.CONTENT_TYPE, MediaType.FORM_DATA.toString())
             .param("testing", "bar")
-            .taskName("my_task_name"));
+            .taskName("my_task_name")
+            .method(HttpMethod.POST));
   }
 
   @Test
@@ -172,7 +134,7 @@ public class BigqueryPollJobActionTest {
     action.run();
     assertLogMessage(
         logHandler, SEVERE, String.format("Bigquery job failed - %s:%s", PROJECT_ID, JOB_ID));
-    assertNoTasksEnqueued(CHAINED_QUEUE_NAME);
+    cloudTasksHelper.assertNoTasksEnqueued(CHAINED_QUEUE_NAME);
   }
 
   @Test
@@ -192,7 +154,7 @@ public class BigqueryPollJobActionTest {
   void testFailure_badChainedTaskPayload() throws Exception {
     when(bigqueryJobsGet.execute()).thenReturn(
         new Job().setStatus(new JobStatus().setState("DONE")));
-    action.payload = "payload".getBytes(UTF_8);
+    action.payload = ByteString.copyFrom("payload".getBytes(UTF_8));
     BadRequestException thrown = assertThrows(BadRequestException.class, action::run);
     assertThat(thrown).hasMessageThat().contains("Cannot deserialize task from payload");
   }

core/src/test/java/google/registry/export/CheckBackupActionTest.java

@@ -15,10 +15,6 @@
 package google.registry.export;
 
 import static com.google.common.truth.Truth.assertThat;
-import static google.registry.export.CheckBackupAction.CHECK_BACKUP_KINDS_TO_LOAD_PARAM;
-import static google.registry.export.CheckBackupAction.CHECK_BACKUP_NAME_PARAM;
-import static google.registry.testing.TaskQueueHelper.assertNoTasksEnqueued;
-import static google.registry.testing.TaskQueueHelper.assertTasksEnqueued;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.Mockito.when;
@@ -27,7 +23,7 @@ import com.google.api.client.googleapis.json.GoogleJsonResponseException;
 import com.google.api.client.http.HttpHeaders;
 import com.google.api.client.json.JsonFactory;
 import com.google.api.client.json.jackson2.JacksonFactory;
-import com.google.common.collect.ImmutableSet;
+import com.google.cloud.tasks.v2.HttpMethod;
 import google.registry.export.datastore.DatastoreAdmin;
 import google.registry.export.datastore.DatastoreAdmin.Get;
 import google.registry.export.datastore.Operation;
@@ -36,9 +32,10 @@ import google.registry.request.HttpException.BadRequestException;
 import google.registry.request.HttpException.NoContentException;
 import google.registry.request.HttpException.NotModifiedException;
 import google.registry.testing.AppEngineExtension;
+import google.registry.testing.CloudTasksHelper;
+import google.registry.testing.CloudTasksHelper.TaskMatcher;
 import google.registry.testing.FakeClock;
 import google.registry.testing.FakeResponse;
-import google.registry.testing.TaskQueueHelper.TaskMatcher;
 import google.registry.testing.TestDataHelper;
 import org.joda.time.DateTime;
 import org.joda.time.Duration;
@@ -57,7 +54,6 @@ public class CheckBackupActionTest {
 
   private static final DateTime START_TIME = DateTime.parse("2014-08-01T01:02:03Z");
   private static final DateTime COMPLETE_TIME = START_TIME.plus(Duration.standardMinutes(30));
-
   private static final JsonFactory JSON_FACTORY = JacksonFactory.getDefaultInstance();
 
   @RegisterExtension
@@ -71,6 +67,7 @@ public class CheckBackupActionTest {
   private final FakeResponse response = new FakeResponse();
   private final FakeClock clock = new FakeClock(COMPLETE_TIME.plusMillis(1000));
   private final CheckBackupAction action = new CheckBackupAction();
+  private final CloudTasksHelper cloudTasksHelper = new CloudTasksHelper();
 
   @BeforeEach
   void beforeEach() throws Exception {
@@ -80,6 +77,7 @@ public class CheckBackupActionTest {
     action.backupName = "some_backup";
     action.kindsToLoadParam = "one,two";
     action.response = response;
+    action.cloudTasksUtils = cloudTasksHelper.getTestCloudTasksUtils();
 
     when(datastoreAdmin.get(anyString())).thenReturn(getBackupProgressRequest);
     when(getBackupProgressRequest.execute()).thenAnswer(arg -> backupOperation);
@@ -110,30 +108,17 @@ public class CheckBackupActionTest {
                 null));
   }
 
-  private static void assertLoadTaskEnqueued(String id, String folder, String kinds) {
-    assertTasksEnqueued(
+  private void assertLoadTaskEnqueued(String id, String folder, String kinds) {
+    cloudTasksHelper.assertTasksEnqueued(
         "export-snapshot",
         new TaskMatcher()
             .url("/_dr/task/uploadDatastoreBackup")
-            .method("POST")
+            .method(HttpMethod.POST)
             .param("id", id)
             .param("folder", folder)
             .param("kinds", kinds));
   }
 
-  @MockitoSettings(strictness = Strictness.LENIENT)
-  @Test
-  void testSuccess_enqueuePollTask() {
-    CheckBackupAction.enqueuePollTask("some_backup_name", ImmutableSet.of("one", "two", "three"));
-    assertTasksEnqueued(
-        CheckBackupAction.QUEUE,
-        new TaskMatcher()
-            .url(CheckBackupAction.PATH)
-            .param(CHECK_BACKUP_NAME_PARAM, "some_backup_name")
-            .param(CHECK_BACKUP_KINDS_TO_LOAD_PARAM, "one,two,three")
-            .method("POST"));
-  }
-
   @Test
   void testPost_forPendingBackup_returnsNotModified() throws Exception {
     setPendingBackup();
@@ -189,7 +174,7 @@ public class CheckBackupActionTest {
     action.kindsToLoadParam = "";
 
     action.run();
-    assertNoTasksEnqueued("export-snapshot");
+    cloudTasksHelper.assertNoTasksEnqueued("export-snapshot");
   }
 
   @MockitoSettings(strictness = Strictness.LENIENT)

core/src/test/java/google/registry/export/UpdateSnapshotViewActionTest.java

@@ -14,15 +14,7 @@
 
 package google.registry.export;
 
-import static com.google.appengine.api.taskqueue.QueueFactory.getQueue;
 import static com.google.common.truth.Truth.assertThat;
-import static google.registry.export.UpdateSnapshotViewAction.QUEUE;
-import static google.registry.export.UpdateSnapshotViewAction.UPDATE_SNAPSHOT_DATASET_ID_PARAM;
-import static google.registry.export.UpdateSnapshotViewAction.UPDATE_SNAPSHOT_KIND_PARAM;
-import static google.registry.export.UpdateSnapshotViewAction.UPDATE_SNAPSHOT_TABLE_ID_PARAM;
-import static google.registry.export.UpdateSnapshotViewAction.UPDATE_SNAPSHOT_VIEWNAME_PARAM;
-import static google.registry.export.UpdateSnapshotViewAction.createViewUpdateTask;
-import static google.registry.testing.TaskQueueHelper.assertTasksEnqueued;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyString;
@@ -38,7 +30,6 @@ import com.google.common.collect.Iterables;
 import google.registry.bigquery.CheckedBigquery;
 import google.registry.request.HttpException.InternalServerErrorException;
 import google.registry.testing.AppEngineExtension;
-import google.registry.testing.TaskQueueHelper.TaskMatcher;
 import java.io.IOException;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -81,23 +72,6 @@ public class UpdateSnapshotViewActionTest {
     action.tableId = "12345_fookind";
   }
 
-  @Test
-  void testSuccess_createViewUpdateTask() {
-    getQueue(QUEUE)
-        .add(
-            createViewUpdateTask(
-                "some_dataset", "12345_fookind", "fookind", "latest_datastore_export"));
-    assertTasksEnqueued(
-        QUEUE,
-        new TaskMatcher()
-            .url(UpdateSnapshotViewAction.PATH)
-            .method("POST")
-            .param(UPDATE_SNAPSHOT_DATASET_ID_PARAM, "some_dataset")
-            .param(UPDATE_SNAPSHOT_TABLE_ID_PARAM, "12345_fookind")
-            .param(UPDATE_SNAPSHOT_KIND_PARAM, "fookind")
-            .param(UPDATE_SNAPSHOT_VIEWNAME_PARAM, "latest_datastore_export"));
-  }
-
   @Test
   void testSuccess_doPost() throws Exception {
     action.run();

core/src/test/java/google/registry/export/UploadDatastoreBackupActionTest.java

@@ -16,16 +16,16 @@ package google.registry.export;
 
 import static com.google.common.collect.Iterables.transform;
 import static com.google.common.truth.Truth.assertThat;
+import static com.google.common.truth.Truth8.assertThat;
+import static google.registry.export.BigqueryPollJobAction.CHAINED_TASK_QUEUE_HEADER;
+import static google.registry.export.BigqueryPollJobAction.JOB_ID_HEADER;
+import static google.registry.export.BigqueryPollJobAction.PROJECT_ID_HEADER;
+import static google.registry.export.UpdateSnapshotViewAction.UPDATE_SNAPSHOT_DATASET_ID_PARAM;
+import static google.registry.export.UpdateSnapshotViewAction.UPDATE_SNAPSHOT_KIND_PARAM;
+import static google.registry.export.UpdateSnapshotViewAction.UPDATE_SNAPSHOT_TABLE_ID_PARAM;
+import static google.registry.export.UpdateSnapshotViewAction.UPDATE_SNAPSHOT_VIEWNAME_PARAM;
 import static google.registry.export.UploadDatastoreBackupAction.BACKUP_DATASET;
-import static google.registry.export.UploadDatastoreBackupAction.LATEST_BACKUP_VIEW_NAME;
-import static google.registry.export.UploadDatastoreBackupAction.PATH;
-import static google.registry.export.UploadDatastoreBackupAction.QUEUE;
-import static google.registry.export.UploadDatastoreBackupAction.UPLOAD_BACKUP_FOLDER_PARAM;
-import static google.registry.export.UploadDatastoreBackupAction.UPLOAD_BACKUP_ID_PARAM;
-import static google.registry.export.UploadDatastoreBackupAction.UPLOAD_BACKUP_KINDS_PARAM;
-import static google.registry.export.UploadDatastoreBackupAction.enqueueUploadBackupTask;
 import static google.registry.export.UploadDatastoreBackupAction.getBackupInfoFileForKind;
-import static google.registry.testing.TaskQueueHelper.assertTasksEnqueued;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.eq;
@@ -38,16 +38,20 @@ import com.google.api.services.bigquery.Bigquery;
 import com.google.api.services.bigquery.model.Dataset;
 import com.google.api.services.bigquery.model.Job;
 import com.google.api.services.bigquery.model.JobConfigurationLoad;
-import com.google.api.services.bigquery.model.JobReference;
-import com.google.appengine.api.taskqueue.QueueFactory;
-import com.google.common.collect.ImmutableSet;
+import com.google.cloud.tasks.v2.HttpMethod;
+import com.google.common.collect.ImmutableMultimap;
 import com.google.common.collect.Iterables;
+import com.google.protobuf.util.Timestamps;
 import google.registry.bigquery.CheckedBigquery;
-import google.registry.export.BigqueryPollJobAction.BigqueryPollJobEnqueuer;
 import google.registry.request.HttpException.InternalServerErrorException;
 import google.registry.testing.AppEngineExtension;
-import google.registry.testing.TaskQueueHelper.TaskMatcher;
+import google.registry.testing.CloudTasksHelper;
+import google.registry.testing.CloudTasksHelper.TaskMatcher;
+import google.registry.testing.FakeClock;
+import google.registry.util.CloudTasksUtils;
+import java.io.ByteArrayInputStream;
 import java.io.IOException;
+import java.io.ObjectInputStream;
 import java.util.List;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -67,8 +71,9 @@ public class UploadDatastoreBackupActionTest {
   private final Bigquery.Datasets bigqueryDatasets = mock(Bigquery.Datasets.class);
   private final Bigquery.Datasets.Insert bigqueryDatasetsInsert =
       mock(Bigquery.Datasets.Insert.class);
-  private final BigqueryPollJobEnqueuer bigqueryPollEnqueuer = mock(BigqueryPollJobEnqueuer.class);
   private UploadDatastoreBackupAction action;
+  private CloudTasksHelper cloudTasksHelper = new CloudTasksHelper();
+  private CloudTasksUtils cloudTasksUtils = cloudTasksHelper.getTestCloudTasksUtils();
 
   @BeforeEach
   void beforeEach() throws Exception {
@@ -80,25 +85,14 @@ public class UploadDatastoreBackupActionTest {
         .thenReturn(bigqueryDatasetsInsert);
     action = new UploadDatastoreBackupAction();
     action.checkedBigquery = checkedBigquery;
-    action.bigqueryPollEnqueuer = bigqueryPollEnqueuer;
     action.projectId = "Project-Id";
     action.backupFolderUrl = "gs://bucket/path";
     action.backupId = "2018-12-05T17:46:39_92612";
     action.backupKinds = "one,two,three";
+    action.cloudTasksUtils = cloudTasksHelper.getTestCloudTasksUtils();
+    action.clock = new FakeClock();
   }
 
-  @Test
-  void testSuccess_enqueueLoadTask() {
-    enqueueUploadBackupTask("id12345", "gs://bucket/path", ImmutableSet.of("one", "two", "three"));
-    assertTasksEnqueued(
-        QUEUE,
-        new TaskMatcher()
-            .url(PATH)
-            .method("POST")
-            .param(UPLOAD_BACKUP_ID_PARAM, "id12345")
-            .param(UPLOAD_BACKUP_FOLDER_PARAM, "gs://bucket/path")
-            .param(UPLOAD_BACKUP_KINDS_PARAM, "one,two,three"));
-  }
 
   @Test
   void testSuccess_doPost() throws Exception {
@@ -153,33 +147,91 @@ public class UploadDatastoreBackupActionTest {
     verify(bigqueryJobsInsert, times(3)).execute();
 
     // Check that the poll tasks for each load job were enqueued.
-    verify(bigqueryPollEnqueuer)
-        .enqueuePollTask(
-            new JobReference()
-                .setProjectId("Project-Id")
-                .setJobId("load-backup-2018_12_05T17_46_39_92612-one"),
-            UpdateSnapshotViewAction.createViewUpdateTask(
-                BACKUP_DATASET, "2018_12_05T17_46_39_92612_one", "one", LATEST_BACKUP_VIEW_NAME),
-            QueueFactory.getQueue(UpdateSnapshotViewAction.QUEUE));
-    verify(bigqueryPollEnqueuer)
-        .enqueuePollTask(
-            new JobReference()
-                .setProjectId("Project-Id")
-                .setJobId("load-backup-2018_12_05T17_46_39_92612-two"),
-            UpdateSnapshotViewAction.createViewUpdateTask(
-                BACKUP_DATASET, "2018_12_05T17_46_39_92612_two", "two", LATEST_BACKUP_VIEW_NAME),
-            QueueFactory.getQueue(UpdateSnapshotViewAction.QUEUE));
-    verify(bigqueryPollEnqueuer)
-        .enqueuePollTask(
-            new JobReference()
-                .setProjectId("Project-Id")
-                .setJobId("load-backup-2018_12_05T17_46_39_92612-three"),
-            UpdateSnapshotViewAction.createViewUpdateTask(
-                BACKUP_DATASET,
-                "2018_12_05T17_46_39_92612_three",
-                "three",
-                LATEST_BACKUP_VIEW_NAME),
-            QueueFactory.getQueue(UpdateSnapshotViewAction.QUEUE));
+    cloudTasksHelper.assertTasksEnqueued(
+        BigqueryPollJobAction.QUEUE,
+        new TaskMatcher()
+            .method(HttpMethod.POST)
+            .header(PROJECT_ID_HEADER, "Project-Id")
+            .header(JOB_ID_HEADER, "load-backup-2018_12_05T17_46_39_92612-one")
+            .header(CHAINED_TASK_QUEUE_HEADER, UpdateSnapshotViewAction.QUEUE)
+            .scheduleTime(
+                Timestamps.fromMillis(
+                    action.clock.nowUtc().plus(BigqueryPollJobAction.POLL_COUNTDOWN).getMillis())),
+        new TaskMatcher()
+            .method(HttpMethod.POST)
+            .header(PROJECT_ID_HEADER, "Project-Id")
+            .header(JOB_ID_HEADER, "load-backup-2018_12_05T17_46_39_92612-two")
+            .header(CHAINED_TASK_QUEUE_HEADER, UpdateSnapshotViewAction.QUEUE)
+            .scheduleTime(
+                Timestamps.fromMillis(
+                    action.clock.nowUtc().plus(BigqueryPollJobAction.POLL_COUNTDOWN).getMillis())),
+        new TaskMatcher()
+            .method(HttpMethod.POST)
+            .header(PROJECT_ID_HEADER, "Project-Id")
+            .header(JOB_ID_HEADER, "load-backup-2018_12_05T17_46_39_92612-three")
+            .header(CHAINED_TASK_QUEUE_HEADER, UpdateSnapshotViewAction.QUEUE)
+            .scheduleTime(
+                Timestamps.fromMillis(
+                    action.clock.nowUtc().plus(BigqueryPollJobAction.POLL_COUNTDOWN).getMillis())));
+
+    // assert the chained task of each enqueud task is correct
+    assertThat(
+            cloudTasksHelper.getTestTasksFor(BigqueryPollJobAction.QUEUE).stream()
+                .map(
+                    testTask -> {
+                      try {
+                        return new ObjectInputStream(
+                                new ByteArrayInputStream(
+                                    testTask.getAppEngineHttpRequest().getBody().toByteArray()))
+                            .readObject();
+                      } catch (ClassNotFoundException | IOException e) {
+                        return null;
+                      }
+                    }))
+        .containsExactly(
+            cloudTasksHelper
+                .getTestCloudTasksUtils()
+                .createPostTask(
+                    UpdateSnapshotViewAction.PATH,
+                    "BACKEND",
+                    ImmutableMultimap.of(
+                        UPDATE_SNAPSHOT_DATASET_ID_PARAM,
+                        "datastore_backups",
+                        UPDATE_SNAPSHOT_TABLE_ID_PARAM,
+                        "2018_12_05T17_46_39_92612_one",
+                        UPDATE_SNAPSHOT_KIND_PARAM,
+                        "one",
+                        UPDATE_SNAPSHOT_VIEWNAME_PARAM,
+                        "latest_datastore_export")),
+            cloudTasksHelper
+                .getTestCloudTasksUtils()
+                .createPostTask(
+                    UpdateSnapshotViewAction.PATH,
+                    "BACKEND",
+                    ImmutableMultimap.of(
+                        UPDATE_SNAPSHOT_DATASET_ID_PARAM,
+                        "datastore_backups",
+                        UPDATE_SNAPSHOT_TABLE_ID_PARAM,
+                        "2018_12_05T17_46_39_92612_two",
+                        UPDATE_SNAPSHOT_KIND_PARAM,
+                        "two",
+                        UPDATE_SNAPSHOT_VIEWNAME_PARAM,
+                        "latest_datastore_export")),
+            cloudTasksHelper
+                .getTestCloudTasksUtils()
+                .createPostTask(
+                    UpdateSnapshotViewAction.PATH,
+                    "BACKEND",
+                    ImmutableMultimap.of(
+                        UPDATE_SNAPSHOT_DATASET_ID_PARAM,
+                        "datastore_backups",
+                        UPDATE_SNAPSHOT_TABLE_ID_PARAM,
+                        "2018_12_05T17_46_39_92612_three",
+                        UPDATE_SNAPSHOT_KIND_PARAM,
+                        "three",
+                        UPDATE_SNAPSHOT_VIEWNAME_PARAM,
+                        "latest_datastore_export")))
+        .inOrder();
   }
 
   @Test

core/src/test/java/google/registry/model/common/ClassPathManagerTest.java

@@ -41,6 +41,7 @@ import google.registry.model.rde.RdeRevision;
 import google.registry.model.registrar.Registrar;
 import google.registry.model.registrar.RegistrarContact;
 import google.registry.model.replay.LastSqlTransaction;
+import google.registry.model.replay.ReplayGap;
 import google.registry.model.reporting.HistoryEntry;
 import google.registry.model.server.Lock;
 import google.registry.model.server.ServerSecret;
@@ -74,6 +75,7 @@ public class ClassPathManagerTest {
     assertThat(ClassPathManager.getClass("HostResource")).isEqualTo(HostResource.class);
     assertThat(ClassPathManager.getClass("Recurring")).isEqualTo(Recurring.class);
     assertThat(ClassPathManager.getClass("Registrar")).isEqualTo(Registrar.class);
+    assertThat(ClassPathManager.getClass("ReplayGap")).isEqualTo(ReplayGap.class);
     assertThat(ClassPathManager.getClass("ContactResource")).isEqualTo(ContactResource.class);
     assertThat(ClassPathManager.getClass("Cancellation")).isEqualTo(Cancellation.class);
     assertThat(ClassPathManager.getClass("RegistrarContact")).isEqualTo(RegistrarContact.class);
@@ -141,6 +143,7 @@ public class ClassPathManagerTest {
     assertThat(ClassPathManager.getClassName(HostResource.class)).isEqualTo("HostResource");
     assertThat(ClassPathManager.getClassName(Recurring.class)).isEqualTo("Recurring");
     assertThat(ClassPathManager.getClassName(Registrar.class)).isEqualTo("Registrar");
+    assertThat(ClassPathManager.getClassName(ReplayGap.class)).isEqualTo("ReplayGap");
     assertThat(ClassPathManager.getClassName(ContactResource.class)).isEqualTo("ContactResource");
     assertThat(ClassPathManager.getClassName(Cancellation.class)).isEqualTo("Cancellation");
     assertThat(ClassPathManager.getClassName(RegistrarContact.class)).isEqualTo("RegistrarContact");

core/src/test/java/google/registry/model/replay/ReplicateToDatastoreActionTest.java

@@ -27,11 +27,13 @@ import static org.junit.jupiter.api.Assumptions.assumeTrue;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
+import com.google.common.base.Supplier;
 import com.google.common.base.Suppliers;
 import com.google.common.collect.ImmutableSortedMap;
 import com.google.common.flogger.FluentLogger;
 import com.google.common.testing.TestLogHandler;
 import com.google.common.truth.Truth8;
+import com.googlecode.objectify.Key;
 import google.registry.model.common.DatabaseMigrationStateSchedule;
 import google.registry.model.common.DatabaseMigrationStateSchedule.MigrationState;
 import google.registry.model.domain.token.AllocationToken;
@@ -39,6 +41,8 @@ import google.registry.model.domain.token.AllocationToken.TokenType;
 import google.registry.model.ofy.CommitLogBucket;
 import google.registry.model.ofy.Ofy;
 import google.registry.model.server.Lock;
+import google.registry.persistence.VKey;
+import google.registry.persistence.transaction.JpaTransactionManagerImpl;
 import google.registry.persistence.transaction.TransactionEntity;
 import google.registry.testing.AppEngineExtension;
 import google.registry.testing.DatabaseHelper;
@@ -48,9 +52,13 @@ import google.registry.testing.InjectExtension;
 import google.registry.testing.ReplayExtension;
 import google.registry.testing.TestObject;
 import google.registry.util.RequestStatusChecker;
+import java.lang.reflect.Proxy;
 import java.util.List;
 import java.util.logging.Level;
 import java.util.logging.Logger;
+import javax.persistence.EntityManager;
+import javax.persistence.EntityManagerFactory;
+import javax.persistence.EntityTransaction;
 import org.joda.time.DateTime;
 import org.joda.time.Duration;
 import org.junit.jupiter.api.AfterEach;
@@ -241,6 +249,99 @@ public class ReplicateToDatastoreActionTest {
     assertThat(ofyTm().transact(() -> LastSqlTransaction.load()).getTransactionId()).isEqualTo(4);
   }
 
+  @Test
+  void testTransactionGapReplay() {
+    insertInDb(TestObject.create("foo"));
+    DeferredCommit deferred = new DeferredCommit(fakeClock);
+    TestObject bar = TestObject.create("bar");
+    deferred.transact(() -> jpaTm().insert(bar));
+    TestObject baz = TestObject.create("baz");
+    insertInDb(baz);
+
+    action.run();
+    assertThat(ofyTm().transact(() -> ofyTm().loadByKeyIfPresent(bar.key())).isPresent()).isFalse();
+    assertThat(ofyTm().transact(() -> ofyTm().loadByKey(baz.key()))).isEqualTo(baz);
+    VKey<ReplayGap> gapKey = VKey.createOfy(ReplayGap.class, Key.create(ReplayGap.class, 2));
+    Truth8.assertThat(ofyTm().transact(() -> ofyTm().loadByKeyIfPresent(gapKey))).isPresent();
+
+    deferred.commit();
+    resetAction();
+    action.run();
+
+    assertThat(ofyTm().transact(() -> ofyTm().loadByKey(bar.key()))).isEqualTo(bar);
+    // Verify that the gap record has been cleaned up.
+    assertThat(ofyTm().transact(() -> ofyTm().loadByKeyIfPresent(gapKey).isPresent())).isFalse();
+  }
+
+  /** Verify that we can handle creation and deletion of > 25 gap records. */
+  @Test
+  void testLargeNumberOfGaps() {
+    // Fail thirty transactions.
+    for (int i = 0; i < 30; ++i) {
+      try {
+        jpaTm()
+            .transact(
+                () -> {
+                  insertInDb(TestObject.create("foo"));
+                  // Explicitly save the transaction entity to force the id update.
+                  jpaTm().insert(new TransactionEntity(new byte[] {1, 2, 3}));
+                  throw new RuntimeException("fail!!!");
+                });
+      } catch (Exception e) {
+        ;
+      }
+    }
+
+    TestObject bar = TestObject.create("bar");
+    insertInDb(bar);
+
+    // Verify that the transaction was successfully applied and that we have generated 30 gap
+    // records.
+    action.run();
+    Truth8.assertThat(ofyTm().transact(() -> ofyTm().loadByKeyIfPresent(bar.key()))).isPresent();
+    assertThat(ofyTm().loadAllOf(ReplayGap.class).size()).isEqualTo(30);
+
+    // Verify that we can clean up this many gap records after expiration.
+    fakeClock.advanceBy(Duration.millis(ReplicateToDatastoreAction.MAX_GAP_RETENTION_MILLIS + 1));
+    resetAction();
+    action.run();
+    assertThat(ofyTm().loadAllOf(ReplayGap.class).size()).isEqualTo(0);
+  }
+
+  @Test
+  void testGapRecordExpiration() {
+    insertInDb(TestObject.create("foo"));
+
+    // Fail a transaction, create a gap.
+    try {
+      jpaTm()
+          .transact(
+              () -> {
+                jpaTm().insert(TestObject.create("other"));
+                // Explicitly save the transaction entity to force the id update.
+                jpaTm().insert(new TransactionEntity(new byte[] {1, 2, 3}));
+                throw new RuntimeException("fail!!!");
+              });
+    } catch (Exception e) {
+      logger.atInfo().log("Got expected exception.");
+    }
+
+    insertInDb(TestObject.create("bar"));
+
+    action.run();
+
+    // Verify that the gap record has been created
+    VKey<ReplayGap> gapKey = VKey.createOfy(ReplayGap.class, Key.create(ReplayGap.class, 2));
+    Truth8.assertThat(ofyTm().transact(() -> ofyTm().loadByKeyIfPresent(gapKey))).isPresent();
+
+    fakeClock.advanceBy(Duration.millis(ReplicateToDatastoreAction.MAX_GAP_RETENTION_MILLIS + 1));
+    resetAction();
+    action.run();
+
+    // Verify that the gap record has been destroyed.
+    assertThat(ofyTm().transact(() -> ofyTm().loadByKeyIfPresent(gapKey)).isPresent()).isFalse();
+  }
+
   @Test
   void testBeforeDatastoreSaveCallback() {
     assumeTrue(ReplayExtension.replayTestsEnabled());
@@ -327,4 +428,63 @@ public class ReplicateToDatastoreActionTest {
     when(requestStatusChecker.getLogId()).thenReturn("logId");
     action = new ReplicateToDatastoreAction(fakeClock, requestStatusChecker, response);
   }
+
+  /**
+   * Deep fake of EntityManagerFactory -> EntityManager -> EntityTransaction that allows us to defer
+   * the actual commit until after the other transactions are replayed.
+   */
+  static class DeferredCommit {
+
+    FakeClock clock;
+    EntityTransaction realTransaction;
+
+    DeferredCommit(FakeClock clock) {
+      this.clock = clock;
+    }
+
+    private static <T> T makeProxy(
+        Class<T> iface, T delegate, String method, Supplier<?> supplier) {
+      return (T)
+          Proxy.newProxyInstance(
+              delegate.getClass().getClassLoader(),
+              new Class[] {iface},
+              (proxy, meth, args) -> {
+                if (meth.getName().equals(method)) {
+                  return supplier.get();
+                } else {
+                  return meth.invoke(delegate, args);
+                }
+              });
+    }
+
+    EntityManager createEntityManagerProxy(EntityManager orgEntityManager) {
+      return makeProxy(
+          EntityManager.class,
+          orgEntityManager,
+          "getTransaction",
+          () ->
+              makeProxy(
+                  EntityTransaction.class,
+                  realTransaction = orgEntityManager.getTransaction(),
+                  "commit",
+                  () -> null));
+    }
+
+    void commit() {
+      realTransaction.commit();
+    }
+
+    void transact(Runnable runnable) {
+      EntityManagerFactory orgEmf =
+          jpaTm().transact(() -> jpaTm().getEntityManager().getEntityManagerFactory());
+      EntityManagerFactory emfProxy =
+          makeProxy(
+              EntityManagerFactory.class,
+              orgEmf,
+              "createEntityManager",
+              () -> createEntityManagerProxy(orgEmf.createEntityManager()));
+
+      new JpaTransactionManagerImpl(emfProxy, clock).transact(runnable);
+    }
+  }
 }

core/src/test/java/google/registry/rde/BrdaCopyActionTest.java

@@ -16,9 +16,13 @@ package google.registry.rde;
 
 import static com.google.common.truth.Truth.assertThat;
 import static com.google.common.truth.Truth.assertWithMessage;
+import static google.registry.model.common.Cursor.CursorType.BRDA;
 import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
+import static google.registry.testing.DatabaseHelper.createTld;
+import static google.registry.testing.DatabaseHelper.persistResource;
 import static google.registry.testing.SystemInfo.hasCommand;
 import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assumptions.assumeTrue;
 
 import com.google.cloud.storage.BlobId;
@@ -28,8 +32,11 @@ import com.google.common.io.CharStreams;
 import com.google.common.io.Files;
 import google.registry.gcs.GcsUtils;
 import google.registry.keyring.api.Keyring;
+import google.registry.model.common.Cursor;
 import google.registry.model.rde.RdeMode;
 import google.registry.model.rde.RdeRevision;
+import google.registry.model.tld.Registry;
+import google.registry.request.HttpException.NoContentException;
 import google.registry.testing.AppEngineExtension;
 import google.registry.testing.BouncyCastleProviderExtension;
 import google.registry.testing.FakeKeyringModule;
@@ -104,6 +111,7 @@ public class BrdaCopyActionTest {
 
   @BeforeEach
   void beforeEach() throws Exception {
+    createTld("lol");
     action.gcsUtils = gcsUtils;
     action.tld = "lol";
     action.watermark = DateTime.parse("2010-10-17TZ");
@@ -116,6 +124,20 @@ public class BrdaCopyActionTest {
             () -> {
               RdeRevision.saveRevision("lol", DateTime.parse("2010-10-17TZ"), RdeMode.THIN, 0);
             });
+    persistResource(Cursor.create(BRDA, action.watermark.plusDays(1), Registry.get("lol")));
+  }
+
+  @ParameterizedTest
+  @ValueSource(strings = {"", "job-name/"})
+  void testRun_stagingNotFinished_throws204(String prefix) throws Exception {
+    persistResource(Cursor.create(BRDA, action.watermark, Registry.get("lol")));
+    NoContentException thrown = assertThrows(NoContentException.class, () -> runAction(prefix));
+    assertThat(thrown)
+        .hasMessageThat()
+        .isEqualTo(
+            "Waiting on RdeStagingAction for TLD lol to copy BRDA deposit for"
+                + " 2010-10-17T00:00:00.000Z to GCS; last BRDA staging completion was before"
+                + " 2010-10-17T00:00:00.000Z");
   }
 
   @ParameterizedTest

core/src/test/java/google/registry/rde/RdeUploadActionTest.java

@@ -403,7 +403,7 @@ public class RdeUploadActionTest {
         .hasMessageThat()
         .isEqualTo(
             "Waiting on RdeStagingAction for TLD tld to send 2010-10-17T00:00:00.000Z upload; last"
-                + " RDE staging completion was at 1970-01-01T00:00:00.000Z");
+                + " RDE staging completion was before 1970-01-01T00:00:00.000Z");
     cloudTasksHelper.assertNoTasksEnqueued("rde-upload");
     assertThat(folder.list()).isEmpty();
   }
@@ -420,7 +420,7 @@ public class RdeUploadActionTest {
         .hasMessageThat()
         .isEqualTo(
             "Waiting on RdeStagingAction for TLD tld to send 2010-10-17T00:00:00.000Z upload; "
-                + "last RDE staging completion was at 2010-10-17T00:00:00.000Z");
+                + "last RDE staging completion was before 2010-10-17T00:00:00.000Z");
   }
 
   @TestOfyAndSql
@@ -437,8 +437,9 @@ public class RdeUploadActionTest {
     assertThat(thrown)
         .hasMessageThat()
         .isEqualTo(
-            "Waiting on 120 minute SFTP cooldown for TLD tld to send 2010-10-17T00:00:00.000Z "
-                + "upload; last upload attempt was at 2010-10-16T22:23:00.000Z (97 minutes ago)");
+            "Waiting on 120 minute SFTP cooldown for TLD tld to send 2010-10-17T00:00:00.000Z"
+                + " upload; last upload attempt was at 2010-10-16T22:23:00.000Z (97 minutes"
+                + " ago)");
   }
 
   private String slurp(InputStream is) throws IOException {

core/src/test/java/google/registry/tools/DomainLockUtilsTest.java

@@ -31,8 +31,6 @@ import static google.registry.tools.LockOrUnlockDomainCommand.REGISTRY_LOCK_STAT
 import static org.joda.time.Duration.standardDays;
 import static org.joda.time.Duration.standardHours;
 import static org.junit.jupiter.api.Assertions.assertThrows;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
 
 import com.google.cloud.tasks.v2.HttpMethod;
 import com.google.common.collect.ImmutableList;
@@ -55,7 +53,6 @@ import google.registry.testing.FakeClock;
 import google.registry.testing.SqlHelper;
 import google.registry.testing.TestOfyAndSql;
 import google.registry.testing.UserInfo;
-import google.registry.util.AppEngineServiceUtils;
 import google.registry.util.StringGenerator.Alphabets;
 import java.util.Optional;
 import java.util.Set;
@@ -96,8 +93,6 @@ public final class DomainLockUtilsTest {
     HostResource host = persistActiveHost("ns1.example.net");
     domain = persistResource(newDomainBase(DOMAIN_NAME, host));
 
-    AppEngineServiceUtils appEngineServiceUtils = mock(AppEngineServiceUtils.class);
-    when(appEngineServiceUtils.getServiceHostname("backend")).thenReturn("backend.hostname.fake");
     domainLockUtils =
         new DomainLockUtils(
             new DeterministicStringGenerator(Alphabets.BASE_58),

core/src/test/java/google/registry/tools/GenerateEscrowDepositCommandTest.java

@@ -17,17 +17,14 @@ package google.registry.tools;
 import static com.google.common.truth.Truth.assertThat;
 import static google.registry.testing.DatabaseHelper.createTld;
 import static org.junit.jupiter.api.Assertions.assertThrows;
-import static org.mockito.Mockito.when;
 
 import com.beust.jcommander.ParameterException;
 import google.registry.testing.CloudTasksHelper;
 import google.registry.testing.CloudTasksHelper.TaskMatcher;
 import google.registry.testing.InjectExtension;
-import google.registry.util.AppEngineServiceUtils;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.RegisterExtension;
-import org.mockito.Mock;
 import org.mockito.junit.jupiter.MockitoSettings;
 import org.mockito.quality.Strictness;
 
@@ -38,8 +35,6 @@ public class GenerateEscrowDepositCommandTest
 
   @RegisterExtension public final InjectExtension inject = new InjectExtension();
 
-  @Mock AppEngineServiceUtils appEngineServiceUtils;
-
   CloudTasksHelper cloudTasksHelper = new CloudTasksHelper();
 
   @BeforeEach
@@ -47,10 +42,7 @@ public class GenerateEscrowDepositCommandTest
     createTld("tld");
     createTld("anothertld");
     command = new GenerateEscrowDepositCommand();
-    command.appEngineServiceUtils = appEngineServiceUtils;
     command.cloudTasksUtils = cloudTasksHelper.getTestCloudTasksUtils();
-    when(appEngineServiceUtils.getCurrentVersionHostname("backend"))
-        .thenReturn("backend.test.localhost");
   }
 
   @Test

core/src/test/java/google/registry/ui/server/registrar/RegistrarSettingsActionTestCase.java

@@ -50,7 +50,6 @@ import google.registry.testing.CloudTasksHelper;
 import google.registry.testing.FakeClock;
 import google.registry.testing.InjectExtension;
 import google.registry.ui.server.SendEmailUtils;
-import google.registry.util.AppEngineServiceUtils;
 import google.registry.util.EmailMessage;
 import google.registry.util.SendEmailService;
 import java.io.PrintWriter;
@@ -88,7 +87,6 @@ public abstract class RegistrarSettingsActionTestCase {
 
   @RegisterExtension public final InjectExtension inject = new InjectExtension();
 
-  @Mock AppEngineServiceUtils appEngineServiceUtils;
   @Mock HttpServletRequest req;
   @Mock HttpServletResponse rsp;
   @Mock SendEmailService emailService;
@@ -112,8 +110,6 @@ public abstract class RegistrarSettingsActionTestCase {
         getOnlyElement(loadRegistrar(CLIENT_ID).getContactsOfType(RegistrarContact.Type.TECH));
 
     action.registrarAccessor = null;
-    action.appEngineServiceUtils = appEngineServiceUtils;
-    when(appEngineServiceUtils.getCurrentVersionHostname("backend")).thenReturn("backend.hostname");
     action.jsonActionRunner =
         new JsonActionRunner(ImmutableMap.of(), new JsonResponse(new ResponseImpl(rsp)));
     action.sendEmailUtils =

core/src/test/resources/google/registry/model/schema.txt

@@ -641,6 +641,10 @@ class google.registry.model.replay.LastSqlTransaction {
   @Id long id;
   long transactionId;
 }
+class google.registry.model.replay.ReplayGap {
+  @Id long transactionId;
+  org.joda.time.DateTime timestamp;
+}
 class google.registry.model.reporting.DomainTransactionRecord {
   google.registry.model.reporting.DomainTransactionRecord$TransactionReportField reportField;
   java.lang.Integer reportAmount;

db/src/main/resources/sql/er_diagram/full_er_diagram.html

@@ -261,11 +261,11 @@ td.section {
     </tr> 
     <tr> 
      <td class="property_name">generated on</td> 
-     <td class="property_value">2022-03-09 22:28:56.561494</td> 
+     <td class="property_value">2022-03-16 18:17:27.007916</td> 
     </tr> 
     <tr>
      <td class="property_name">last flyway file</td>
-     <td id="lastFlywayFile" class="property_value">V112__add_billingrecurrence_missing_indexes.sql</td>
+     <td id="lastFlywayFile" class="property_value">V114__add_allocation_token_indexes.sql</td>
     </tr>
    </tbody>
   </table> 
@@ -284,7 +284,7 @@ td.section {
      generated on
     </text> 
     <text text-anchor="start" x="4755.52" y="-10.8" font-family="Helvetica,sans-Serif" font-size="14.00">
-     2022-03-09 22:28:56.561494
+     2022-03-16 18:17:27.007916
     </text> 
     <polygon fill="none" stroke="#888888" points="4668.02,-4 4668.02,-44 4933.02,-44 4933.02,-4 4668.02,-4" /> <!-- allocationtoken_a08ccbef --> 
     <g id="node1" class="node"> 
@@ -6698,6 +6698,18 @@ td.section {
     <tr> 
      <td colspan="3"></td> 
     </tr> 
+    <tr> 
+     <td colspan="2" class="name">idxtmlqd31dpvvd2g1h9i7erw6aj</td> 
+     <td class="description right">[non-unique index]</td> 
+    </tr> 
+    <tr> 
+     <td class="spacer"></td> 
+     <td class="minwidth">redemption_domain_repo_id</td> 
+     <td class="minwidth">ascending</td> 
+    </tr> 
+    <tr> 
+     <td colspan="3"></td> 
+    </tr> 
     <tr> 
      <td colspan="2" class="name">AllocationToken_pkey</td> 
      <td class="description right">[unique index]</td> 
@@ -6707,6 +6719,18 @@ td.section {
      <td class="minwidth">token</td> 
      <td class="minwidth">ascending</td> 
     </tr> 
+    <tr> 
+     <td colspan="3"></td> 
+    </tr> 
+    <tr> 
+     <td colspan="2" class="name">idx9g3s7mjv1yn4t06nqid39whss</td> 
+     <td class="description right">[non-unique index]</td> 
+    </tr> 
+    <tr> 
+     <td class="spacer"></td> 
+     <td class="minwidth">token_type</td> 
+     <td class="minwidth">ascending</td> 
+    </tr> 
    </tbody>
   </table> 
   <p>&nbsp;</p> 
@@ -11323,6 +11347,18 @@ td.section {
     <tr> 
      <td colspan="3"></td> 
     </tr> 
+    <tr> 
+     <td colspan="2" class="name">idxl49vydnq0h5j1piefwjy4i8er</td> 
+     <td class="description right">[non-unique index]</td> 
+    </tr> 
+    <tr> 
+     <td class="spacer"></td> 
+     <td class="minwidth">current_sponsor_registrar_id</td> 
+     <td class="minwidth">ascending</td> 
+    </tr> 
+    <tr> 
+     <td colspan="3"></td> 
+    </tr> 
     <tr> 
      <td colspan="2" class="name">idxkpkh68n6dy5v51047yr6b0e9l</td> 
      <td class="description right">[non-unique index]</td> 
@@ -11344,6 +11380,30 @@ td.section {
      <td class="minwidth">repo_id</td> 
      <td class="minwidth">ascending</td> 
     </tr> 
+    <tr> 
+     <td colspan="3"></td> 
+    </tr> 
+    <tr> 
+     <td colspan="2" class="name">idxy98mebut8ix1v07fjxxdkqcx</td> 
+     <td class="description right">[non-unique index]</td> 
+    </tr> 
+    <tr> 
+     <td class="spacer"></td> 
+     <td class="minwidth">creation_time</td> 
+     <td class="minwidth">ascending</td> 
+    </tr> 
+    <tr> 
+     <td colspan="3"></td> 
+    </tr> 
+    <tr> 
+     <td colspan="2" class="name">idxovmntef6l45tw2bsfl56tcugx</td> 
+     <td class="description right">[non-unique index]</td> 
+    </tr> 
+    <tr> 
+     <td class="spacer"></td> 
+     <td class="minwidth">deletion_time</td> 
+     <td class="minwidth">ascending</td> 
+    </tr> 
    </tbody>
   </table> 
   <p>&nbsp;</p> 

db/src/main/resources/sql/flyway.txt

@@ -110,3 +110,5 @@ V109__add_domain_host_missing_indexes_from_query_analyzer.sql
 V110__add_graceperiod_missing_indexes_from_query_analyzer.sql
 V111__add_billingcancellation_missing_indexes.sql
 V112__add_billingrecurrence_missing_indexes.sql
+V113__add_host_missing_indexes.sql
+V114__add_allocation_token_indexes.sql

db/src/main/resources/sql/flyway/V113__add_host_missing_indexes.sql

@@ -0,0 +1,18 @@
+-- Copyright 2022 The Nomulus Authors. All Rights Reserved.
+--
+-- 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.
+
+CREATE INDEX IDXy98mebut8ix1v07fjxxdkqcx ON "Host" (creation_time);
+CREATE INDEX IDXovmntef6l45tw2bsfl56tcugx ON "Host" (deletion_time);
+CREATE INDEX IDXl49vydnq0h5j1piefwjy4i8er ON "Host" (current_sponsor_registrar_id);
+

db/src/main/resources/sql/flyway/V114__add_allocation_token_indexes.sql

@@ -0,0 +1,17 @@
+-- Copyright 2022 The Nomulus Authors. All Rights Reserved.
+--
+-- 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.
+
+
+CREATE INDEX IDX9g3s7mjv1yn4t06nqid39whss ON "AllocationToken" (token_type);
+CREATE INDEX IDXtmlqd31dpvvd2g1h9i7erw6aj ON "AllocationToken" (redemption_domain_repo_id);

db/src/main/resources/sql/schema/db-schema.sql.generated

@@ -759,16 +759,24 @@
         primary key (id)
     );
 create index allocation_token_domain_name_idx on "AllocationToken" (domain_name);
+create index IDX9g3s7mjv1yn4t06nqid39whss on "AllocationToken" (token_type);
+create index IDXtmlqd31dpvvd2g1h9i7erw6aj on "AllocationToken" (redemption_domain_repo_id);
 create index IDXih4b2tea127p5rb61gje6e1y2 on "BillingCancellation" (registrar_id);
 create index IDX2exdfbx6oiiwnhr8j6gjpqt2j on "BillingCancellation" (event_time);
+create index IDXl8vobbecsd32k4ksavdfx8st6 on "BillingCancellation" (domain_repo_id);
 create index IDXqa3g92jc17e8dtiaviy4fet4x on "BillingCancellation" (billing_time);
+create index IDX4ytbe5f3b39trsd4okx5ijhs4 on "BillingCancellation" (billing_event_id);
+create index IDXku0fopwyvd57ebo8bf0jg9xo2 on "BillingCancellation" (billing_recurrence_id);
 create index IDXqspv57gj2led8ly42fq01t7m7 on "BillingEvent" (registrar_id);
 create index IDX5yfbr88439pxw0v3j86c74fp8 on "BillingEvent" (event_time);
 create index IDX6py6ocrab0ivr76srcd2okpnq on "BillingEvent" (billing_time);
 create index IDXplxf9v56p0wg8ws6qsvd082hk on "BillingEvent" (synthetic_creation_time);
+create index IDXbgfmveqa7e5hn689koikwn70r on "BillingEvent" (domain_repo_id);
 create index IDXcesda59ssop44kklytpb292hn on "BillingEvent" (allocation_token);
+create index IDX6ebt3nwk5ocvnremnhnlkl6ff on "BillingEvent" (cancellation_matching_billing_recurrence_id);
 create index IDXd3gxhkh0jk694pjvh9pyn7wjc on "BillingRecurrence" (registrar_id);
 create index IDX6syykou4nkc7hqa5p8r92cpch on "BillingRecurrence" (event_time);
+create index IDXoqttafcywwdn41um6kwlt0n8b on "BillingRecurrence" (domain_repo_id);
 create index IDXp3usbtvk0v1m14i5tdp4xnxgc on "BillingRecurrence" (recurrence_end_time);
 create index IDXjny8wuot75b5e6p38r47wdawu on "BillingRecurrence" (recurrence_time_of_year);
 create index IDX3y752kr9uh4kh6uig54vemx0l on "Contact" (creation_time);
@@ -792,6 +800,9 @@ create index IDXr22ciyccwi9rrqmt1ro0s59qf on "Domain" (tech_contact);
 create index IDXrwl38wwkli1j7gkvtywi9jokq on "Domain" (tld);
 create index IDXa7fu0bqynfb79rr80528b4jqt on "Domain" (registrant_contact);
 create index IDXcws5mvmpl8o10wrhde780ors2 on "Domain" (dns_refresh_request_time);
+create index IDXsfci08jgsymxy6ovh4k7r358c on "Domain" (billing_recurrence_id);
+create index IDX3y3k7m2bkgahm9sixiohgyrga on "Domain" (transfer_billing_event_id);
+create index IDXcju58vqascbpve1t7fem53ctl on "Domain" (transfer_billing_recurrence_id);
 create index IDXrh4xmrot9bd63o382ow9ltfig on "DomainHistory" (creation_time);
 create index IDXaro1omfuaxjwmotk3vo00trwm on "DomainHistory" (history_registrar_id);
 create index IDXsu1nam10cjes9keobapn5jvxj on "DomainHistory" (history_type);
@@ -799,12 +810,18 @@ create index IDX6w3qbtgce93cal2orjg1tw7b7 on "DomainHistory" (history_modificati
 
     alter table if exists "DomainHistoryHost" 
        add constraint UKt2e7ae3t8gcsxd13wjx2ka7ij unique (domain_history_history_revision_id, domain_history_domain_repo_id, host_repo_id);
+create index IDXjw3rwtfrexyq53x9vu7qghrdt on "DomainHost" (host_repo_id);
 
     alter table if exists "DomainHost" 
        add constraint UKat9erbh52e4lg3jw6ai9wkjj9 unique (domain_repo_id, host_repo_id);
 create index IDXj1mtx98ndgbtb1bkekahms18w on "GracePeriod" (domain_repo_id);
+create index IDXbgssjudpm428mrv0xfpvgifps on "GracePeriod" (billing_event_id);
+create index IDX5u5m6clpk3nktrvtyy5umacb6 on "GracePeriod" (billing_recurrence_id);
 create index IDXd01j17vrpjxaerxdmn8bwxs7s on "GracePeriodHistory" (domain_repo_id);
 create index IDXkpkh68n6dy5v51047yr6b0e9l on "Host" (host_name);
+create index IDXy98mebut8ix1v07fjxxdkqcx on "Host" (creation_time);
+create index IDXovmntef6l45tw2bsfl56tcugx on "Host" (deletion_time);
+create index IDXl49vydnq0h5j1piefwjy4i8er on "Host" (current_sponsor_registrar_id);
 create index IDXfg2nnjlujxo6cb9fha971bq2n on "HostHistory" (creation_time);
 create index IDX1iy7njgb7wjmj9piml4l2g0qi on "HostHistory" (history_registrar_id);
 create index IDXkkwbwcwvrdkkqothkiye4jiff on "HostHistory" (host_name);

db/src/main/resources/sql/schema/nomulus.golden.sql

@@ -1623,6 +1623,13 @@ CREATE INDEX idx73l103vc5900ig3p4odf0cngt ON public."BillingEvent" USING btree (
 CREATE INDEX idx8nr0ke9mrrx4ewj6pd2ag4rmr ON public."Domain" USING btree (creation_time);
 
 
+--
+-- Name: idx9g3s7mjv1yn4t06nqid39whss; Type: INDEX; Schema: public; Owner: -
+--
+
+CREATE INDEX idx9g3s7mjv1yn4t06nqid39whss ON public."AllocationToken" USING btree (token_type);
+
+
 --
 -- Name: idx9q53px6r302ftgisqifmc6put; Type: INDEX; Schema: public; Owner: -
 --
@@ -1798,6 +1805,13 @@ CREATE INDEX idxkpkh68n6dy5v51047yr6b0e9l ON public."Host" USING btree (host_nam
 CREATE INDEX idxku0fopwyvd57ebo8bf0jg9xo2 ON public."BillingCancellation" USING btree (billing_recurrence_id);
 
 
+--
+-- Name: idxl49vydnq0h5j1piefwjy4i8er; Type: INDEX; Schema: public; Owner: -
+--
+
+CREATE INDEX idxl49vydnq0h5j1piefwjy4i8er ON public."Host" USING btree (current_sponsor_registrar_id);
+
+
 --
 -- Name: idxl8vobbecsd32k4ksavdfx8st6; Type: INDEX; Schema: public; Owner: -
 --
@@ -1854,6 +1868,13 @@ CREATE INDEX idxoqd7n4hbx86hvlgkilq75olas ON public."Contact" USING btree (conta
 CREATE INDEX idxoqttafcywwdn41um6kwlt0n8b ON public."BillingRecurrence" USING btree (domain_repo_id);
 
 
+--
+-- Name: idxovmntef6l45tw2bsfl56tcugx; Type: INDEX; Schema: public; Owner: -
+--
+
+CREATE INDEX idxovmntef6l45tw2bsfl56tcugx ON public."Host" USING btree (deletion_time);
+
+
 --
 -- Name: idxp3usbtvk0v1m14i5tdp4xnxgc; Type: INDEX; Schema: public; Owner: -
 --
@@ -1931,6 +1952,20 @@ CREATE INDEX idxsu1nam10cjes9keobapn5jvxj ON public."DomainHistory" USING btree
 CREATE INDEX idxsudwswtwqnfnx2o1hx4s0k0g5 ON public."ContactHistory" USING btree (history_modification_time);
 
 
+--
+-- Name: idxtmlqd31dpvvd2g1h9i7erw6aj; Type: INDEX; Schema: public; Owner: -
+--
+
+CREATE INDEX idxtmlqd31dpvvd2g1h9i7erw6aj ON public."AllocationToken" USING btree (redemption_domain_repo_id);
+
+
+--
+-- Name: idxy98mebut8ix1v07fjxxdkqcx; Type: INDEX; Schema: public; Owner: -
+--
+
+CREATE INDEX idxy98mebut8ix1v07fjxxdkqcx ON public."Host" USING btree (creation_time);
+
+
 --
 -- Name: premiumlist_name_idx; Type: INDEX; Schema: public; Owner: -
 --