Use CloudTasksUtils to enqueue in RegistrarSettingsAction (#1467)

* Use CloudTaskUtils to enqueue

* Add CloudTasksUtilsModule to FrontendComponent

* Fix Uri query issue

* Remove header and check service in matcher

* Use a ThreadLocal boolean in TestServer to determine enqueueing

* Extract enqueuing and email sending from tm().transact()

common/src/main/java/google/registry/util/Sleeper.java

@@ -41,4 +41,20 @@ public interface Sleeper {
    * @see com.google.common.util.concurrent.Uninterruptibles#sleepUninterruptibly
    */
   void sleepUninterruptibly(ReadableDuration duration);
+
+  /**
+   * Puts the current thread to interruptible sleep.
+   *
+   * <p>This is a convenience method for {@link #sleep} that properly converts an {@link
+   * InterruptedException} to a {@link RuntimeException}.
+   */
+  default void sleepInterruptibly(ReadableDuration duration) {
+    try {
+      sleep(duration);
+    } catch (InterruptedException e) {
+      // Restore current thread's interrupted state.
+      Thread.currentThread().interrupt();
+      throw new RuntimeException("Interrupted.", e);
+    }
+  }
 }

core/build.gradle

@@ -319,6 +319,7 @@ dependencies {
   testCompile deps['com.google.appengine:appengine-testing']
   testCompile deps['com.google.guava:guava-testlib']
   testCompile deps['com.google.monitoring-client:contrib']
+  testCompile deps['com.google.protobuf:protobuf-java-util']
   testCompile deps['com.google.truth:truth']
   testCompile deps['com.google.truth.extensions:truth-java8-extension']
   testCompile deps['org.checkerframework:checker-qual']
@@ -676,9 +677,9 @@ Optional<List<String>> getToolArgsList() {
 
 // To run the nomulus tools with these command line tokens:
 // "--foo", "bar baz", "--qux=quz"
-//   gradle registryTool --args="--foo 'bar baz' --qux=quz"
+//   gradle core:registryTool --args="--foo 'bar baz' --qux=quz"
 // or:
-//   gradle registryTool --PtoolArgs="--foo|bar baz|--qux=quz"
+//   gradle core:registryTool -PtoolArgs="--foo|bar baz|--qux=quz"
 // Note that the delimiting pipe can be backslash escaped if it is part of a
 // parameter.
 ext.createToolTask = {
@@ -708,6 +709,9 @@ createToolTask(
 createToolTask(
     'validateSqlPipeline', 'google.registry.beam.comparedb.ValidateSqlPipeline')
 
+createToolTask(
+    'validateDatastorePipeline', 'google.registry.beam.comparedb.ValidateDatastorePipeline')
+
 
 createToolTask(
     'jpaDemoPipeline', 'google.registry.beam.common.JpaDemoPipeline')
@@ -793,6 +797,16 @@ if (environment == 'alpha') {
               mainClass: 'google.registry.beam.rde.RdePipeline',
               metaData : 'google/registry/beam/rde_pipeline_metadata.json'
           ],
+      validateDatastore  :
+          [
+              mainClass: 'google.registry.beam.comparedb.ValidateDatastorePipeline',
+              metaData: 'google/registry/beam/validate_datastore_pipeline_metadata.json'
+          ],
+      validateSql  :
+          [
+              mainClass: 'google.registry.beam.comparedb.ValidateSqlPipeline',
+              metaData: 'google/registry/beam/validate_sql_pipeline_metadata.json'
+          ],
   ]
   project.tasks.create("stageBeamPipelines") {
     doLast {

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

@@ -21,6 +21,7 @@ import static google.registry.backup.ExportCommitLogDiffAction.UPPER_CHECKPOINT_
 import static google.registry.backup.RestoreCommitLogsAction.BUCKET_OVERRIDE_PARAM;
 import static google.registry.backup.RestoreCommitLogsAction.FROM_TIME_PARAM;
 import static google.registry.backup.RestoreCommitLogsAction.TO_TIME_PARAM;
+import static google.registry.backup.SyncDatastoreToSqlSnapshotAction.SQL_SNAPSHOT_ID_PARAM;
 import static google.registry.request.RequestParameters.extractOptionalParameter;
 import static google.registry.request.RequestParameters.extractRequiredDatetimeParameter;
 import static google.registry.request.RequestParameters.extractRequiredParameter;
@@ -98,6 +99,12 @@ public final class BackupModule {
     return extractRequiredDatetimeParameter(req, TO_TIME_PARAM);
   }
 
+  @Provides
+  @Parameter(SQL_SNAPSHOT_ID_PARAM)
+  static String provideSqlSnapshotId(HttpServletRequest req) {
+    return extractRequiredParameter(req, SQL_SNAPSHOT_ID_PARAM);
+  }
+
   @Provides
   @Backups
   static ListeningExecutorService provideListeningExecutorService() {

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

@@ -17,7 +17,7 @@ package google.registry.backup;
 import static google.registry.backup.ExportCommitLogDiffAction.LOWER_CHECKPOINT_TIME_PARAM;
 import static google.registry.backup.ExportCommitLogDiffAction.UPPER_CHECKPOINT_TIME_PARAM;
 import static google.registry.model.ofy.ObjectifyService.auditedOfy;
-import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
+import static google.registry.persistence.transaction.TransactionManagerFactory.ofyTm;
 import static google.registry.util.DateTimeUtils.isBeforeOrAt;
 
 import com.google.common.collect.ImmutableMultimap;
@@ -30,6 +30,7 @@ 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;
 
@@ -64,32 +65,47 @@ public final class CommitLogCheckpointAction implements Runnable {
 
   @Override
   public void run() {
+    createCheckPointAndStartAsyncExport();
+  }
+
+  /**
+   * Creates a {@link CommitLogCheckpoint} and initiates an asynchronous export task.
+   *
+   * @return the {@code CommitLogCheckpoint} to be exported
+   */
+  public Optional<CommitLogCheckpoint> createCheckPointAndStartAsyncExport() {
     final CommitLogCheckpoint checkpoint = strategy.computeCheckpoint();
     logger.atInfo().log(
         "Generated candidate checkpoint for time: %s", checkpoint.getCheckpointTime());
-    tm().transact(
-            () -> {
-              DateTime lastWrittenTime = CommitLogCheckpointRoot.loadRoot().getLastWrittenTime();
-              if (isBeforeOrAt(checkpoint.getCheckpointTime(), lastWrittenTime)) {
-                logger.atInfo().log(
-                    "Newer checkpoint already written at time: %s", lastWrittenTime);
-                return;
-              }
-              auditedOfy()
-                  .saveIgnoringReadOnlyWithoutBackup()
-                  .entities(
-                      checkpoint, CommitLogCheckpointRoot.create(checkpoint.getCheckpointTime()));
-              // Enqueue a diff task between previous and current checkpoints.
-              cloudTasksUtils.enqueue(
-                  QUEUE_NAME,
-                  CloudTasksUtils.createPostTask(
-                      ExportCommitLogDiffAction.PATH,
-                      Service.BACKEND.toString(),
-                      ImmutableMultimap.of(
-                          LOWER_CHECKPOINT_TIME_PARAM,
-                          lastWrittenTime.toString(),
-                          UPPER_CHECKPOINT_TIME_PARAM,
-                          checkpoint.getCheckpointTime().toString())));
-            });
+    boolean isCheckPointPersisted =
+        ofyTm()
+            .transact(
+                () -> {
+                  DateTime lastWrittenTime =
+                      CommitLogCheckpointRoot.loadRoot().getLastWrittenTime();
+                  if (isBeforeOrAt(checkpoint.getCheckpointTime(), lastWrittenTime)) {
+                    logger.atInfo().log(
+                        "Newer checkpoint already written at time: %s", lastWrittenTime);
+                    return false;
+                  }
+                  auditedOfy()
+                      .saveIgnoringReadOnlyWithoutBackup()
+                      .entities(
+                          checkpoint,
+                          CommitLogCheckpointRoot.create(checkpoint.getCheckpointTime()));
+                  // Enqueue a diff task between previous and current checkpoints.
+                  cloudTasksUtils.enqueue(
+                      QUEUE_NAME,
+                      CloudTasksUtils.createPostTask(
+                          ExportCommitLogDiffAction.PATH,
+                          Service.BACKEND.toString(),
+                          ImmutableMultimap.of(
+                              LOWER_CHECKPOINT_TIME_PARAM,
+                              lastWrittenTime.toString(),
+                              UPPER_CHECKPOINT_TIME_PARAM,
+                              checkpoint.getCheckpointTime().toString())));
+                  return true;
+                });
+    return isCheckPointPersisted ? Optional.of(checkpoint) : Optional.empty();
   }
 }

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

@@ -18,7 +18,7 @@ import static com.google.common.base.Preconditions.checkNotNull;
 import static com.google.common.base.Preconditions.checkState;
 import static google.registry.mapreduce.MapreduceRunner.PARAM_DRY_RUN;
 import static google.registry.model.ofy.ObjectifyService.auditedOfy;
-import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
+import static google.registry.persistence.transaction.TransactionManagerFactory.ofyTm;
 import static java.lang.Boolean.FALSE;
 import static java.lang.Boolean.TRUE;
 
@@ -288,7 +288,8 @@ public final class DeleteOldCommitLogsAction implements Runnable {
       }
 
       DeletionResult deletionResult =
-          tm().transactNew(
+          ofyTm()
+              .transactNew(
                   () -> {
                     CommitLogManifest manifest = auditedOfy().load().key(manifestKey).now();
                     // It is possible that the same manifestKey was run twice, if a shard had to be

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

@@ -0,0 +1,173 @@
+// 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.backup;
+
+import static javax.servlet.http.HttpServletResponse.SC_INTERNAL_SERVER_ERROR;
+import static javax.servlet.http.HttpServletResponse.SC_OK;
+
+import com.google.common.flogger.FluentLogger;
+import google.registry.beam.comparedb.LatestDatastoreSnapshotFinder;
+import google.registry.config.RegistryConfig.Config;
+import google.registry.model.annotations.DeleteAfterMigration;
+import google.registry.model.ofy.CommitLogCheckpoint;
+import google.registry.model.replay.ReplicateToDatastoreAction;
+import google.registry.request.Action;
+import google.registry.request.Action.Service;
+import google.registry.request.Parameter;
+import google.registry.request.Response;
+import google.registry.request.auth.Auth;
+import google.registry.util.Sleeper;
+import java.util.Optional;
+import javax.inject.Inject;
+import org.joda.time.DateTime;
+import org.joda.time.Duration;
+
+/**
+ * Synchronizes Datastore to a given SQL snapshot when SQL is the primary database.
+ *
+ * <p>The caller takes the responsibility for:
+ *
+ * <ul>
+ *   <li>verifying the current migration stage
+ *   <li>acquiring the {@link ReplicateToDatastoreAction#REPLICATE_TO_DATASTORE_LOCK_NAME
+ *       replication lock}, and
+ *   <li>while holding the lock, creating an SQL snapshot and invoking this action with the snapshot
+ *       id
+ * </ul>
+ *
+ * The caller may release the replication lock upon receiving the response from this action. Please
+ * refer to {@link google.registry.tools.ValidateDatastoreWithSqlCommand} for more information on
+ * usage.
+ *
+ * <p>This action plays SQL transactions up to the user-specified snapshot, creates a new CommitLog
+ * checkpoint, and exports all CommitLogs to GCS up to this checkpoint. The timestamp of this
+ * checkpoint can be used to recreate a Datastore snapshot that is equivalent to the given SQL
+ * snapshot. If this action succeeds, the checkpoint timestamp is included in the response (the
+ * format of which is defined by {@link #SUCCESS_RESPONSE_TEMPLATE}).
+ */
+@Action(
+    service = Service.BACKEND,
+    path = SyncDatastoreToSqlSnapshotAction.PATH,
+    method = Action.Method.POST,
+    auth = Auth.AUTH_INTERNAL_OR_ADMIN)
+@DeleteAfterMigration
+public class SyncDatastoreToSqlSnapshotAction implements Runnable {
+  private static final FluentLogger logger = FluentLogger.forEnclosingClass();
+
+  public static final String PATH = "/_dr/task/syncDatastoreToSqlSnapshot";
+
+  public static final String SUCCESS_RESPONSE_TEMPLATE =
+      "Datastore is up-to-date with provided SQL snapshot (%s). CommitLog timestamp is (%s).";
+
+  static final String SQL_SNAPSHOT_ID_PARAM = "sqlSnapshotId";
+
+  private static final int COMMITLOGS_PRESENCE_CHECK_ATTEMPTS = 10;
+  private static final Duration COMMITLOGS_PRESENCE_CHECK_DELAY = Duration.standardSeconds(6);
+
+  private final Response response;
+  private final Sleeper sleeper;
+
+  @Config("commitLogGcsBucket")
+  private final String gcsBucket;
+
+  private final GcsDiffFileLister gcsDiffFileLister;
+  private final LatestDatastoreSnapshotFinder datastoreSnapshotFinder;
+  private final CommitLogCheckpointAction commitLogCheckpointAction;
+  private final String sqlSnapshotId;
+
+  @Inject
+  SyncDatastoreToSqlSnapshotAction(
+      Response response,
+      Sleeper sleeper,
+      @Config("commitLogGcsBucket") String gcsBucket,
+      GcsDiffFileLister gcsDiffFileLister,
+      LatestDatastoreSnapshotFinder datastoreSnapshotFinder,
+      CommitLogCheckpointAction commitLogCheckpointAction,
+      @Parameter(SQL_SNAPSHOT_ID_PARAM) String sqlSnapshotId) {
+    this.response = response;
+    this.sleeper = sleeper;
+    this.gcsBucket = gcsBucket;
+    this.gcsDiffFileLister = gcsDiffFileLister;
+    this.datastoreSnapshotFinder = datastoreSnapshotFinder;
+    this.commitLogCheckpointAction = commitLogCheckpointAction;
+    this.sqlSnapshotId = sqlSnapshotId;
+  }
+
+  @Override
+  public void run() {
+    logger.atInfo().log("Datastore validation invoked. SqlSnapshotId is %s.", sqlSnapshotId);
+
+    try {
+      CommitLogCheckpoint checkpoint = ensureDatabasesComparable(sqlSnapshotId);
+      response.setStatus(SC_OK);
+      response.setPayload(
+          String.format(SUCCESS_RESPONSE_TEMPLATE, sqlSnapshotId, checkpoint.getCheckpointTime()));
+      return;
+    } catch (Exception e) {
+      response.setStatus(SC_INTERNAL_SERVER_ERROR);
+      response.setPayload(e.getMessage());
+    }
+  }
+
+  private CommitLogCheckpoint ensureDatabasesComparable(String sqlSnapshotId) {
+    // Replicate SQL transaction to Datastore, up to when this snapshot is taken.
+    int playbacks = ReplicateToDatastoreAction.replayAllTransactions(Optional.of(sqlSnapshotId));
+    logger.atInfo().log("Played %s SQL transactions.", playbacks);
+
+    Optional<CommitLogCheckpoint> checkpoint = exportCommitLogs();
+    if (!checkpoint.isPresent()) {
+      throw new RuntimeException("Cannot create CommitLog checkpoint");
+    }
+    logger.atInfo().log(
+        "CommitLog checkpoint created at %s.", checkpoint.get().getCheckpointTime());
+    verifyCommitLogsPersisted(checkpoint.get());
+    return checkpoint.get();
+  }
+
+  private Optional<CommitLogCheckpoint> exportCommitLogs() {
+    // Trigger an async CommitLog export to GCS. Will check file availability later.
+    // Although we can add support to synchronous execution, it can disrupt the export cadence
+    // when the system is busy
+    Optional<CommitLogCheckpoint> checkpoint =
+        commitLogCheckpointAction.createCheckPointAndStartAsyncExport();
+
+    // Failure to create checkpoint most likely caused by race with cron-triggered checkpointing.
+    // Retry once.
+    if (!checkpoint.isPresent()) {
+      commitLogCheckpointAction.createCheckPointAndStartAsyncExport();
+    }
+    return checkpoint;
+  }
+
+  private void verifyCommitLogsPersisted(CommitLogCheckpoint checkpoint) {
+    DateTime exportStartTime =
+        datastoreSnapshotFinder
+            .getSnapshotInfo(checkpoint.getCheckpointTime().toInstant())
+            .exportInterval()
+            .getStart();
+    logger.atInfo().log("Found Datastore export at %s", exportStartTime);
+    for (int attempts = 0; attempts < COMMITLOGS_PRESENCE_CHECK_ATTEMPTS; attempts++) {
+      try {
+        gcsDiffFileLister.listDiffFiles(gcsBucket, exportStartTime, checkpoint.getCheckpointTime());
+        return;
+      } catch (IllegalStateException e) {
+        // Gap in commitlog files. Fall through to sleep and retry.
+        logger.atInfo().log("Commitlog files not yet found on GCS.");
+      }
+      sleeper.sleepInterruptibly(COMMITLOGS_PRESENCE_CHECK_DELAY);
+    }
+    throw new RuntimeException("Cannot find all commitlog files.");
+  }
+}

core/src/main/java/google/registry/batch/AsyncTaskEnqueuer.java

@@ -110,7 +110,7 @@ public final class AsyncTaskEnqueuer {
             .method(Method.POST)
             .header("Host", backendHostname)
             .countdownMillis(etaDuration.getMillis())
-            .param(PARAM_RESOURCE_KEY, entityKey.getOfyKey().getString())
+            .param(PARAM_RESOURCE_KEY, entityKey.stringify())
             .param(PARAM_REQUESTED_TIME, now.toString());
     if (whenToResave.size() > 1) {
       task.param(PARAM_RESAVE_TIMES, Joiner.on(',').join(whenToResave.tailSet(firstResave, false)));
@@ -131,7 +131,7 @@ public final class AsyncTaskEnqueuer {
     TaskOptions task =
         TaskOptions.Builder.withMethod(Method.PULL)
             .countdownMillis(asyncDeleteDelay.getMillis())
-            .param(PARAM_RESOURCE_KEY, resourceToDelete.createVKey().getOfyKey().getString())
+            .param(PARAM_RESOURCE_KEY, resourceToDelete.createVKey().stringify())
             .param(PARAM_REQUESTING_CLIENT_ID, requestingRegistrarId)
             .param(PARAM_SERVER_TRANSACTION_ID, trid.getServerTransactionId())
             .param(PARAM_IS_SUPERUSER, Boolean.toString(isSuperuser))
@@ -148,7 +148,7 @@ public final class AsyncTaskEnqueuer {
     addTaskToQueueWithRetry(
         asyncDnsRefreshPullQueue,
         TaskOptions.Builder.withMethod(Method.PULL)
-            .param(PARAM_HOST_KEY, hostKey.getOfyKey().getString())
+            .param(PARAM_HOST_KEY, hostKey.stringify())
             .param(PARAM_REQUESTED_TIME, now.toString()));
   }
 

core/src/main/java/google/registry/beam/common/RegistryPipelineComponent.java

@@ -23,6 +23,7 @@ import google.registry.config.RegistryConfig.ConfigModule;
 import google.registry.persistence.PersistenceModule;
 import google.registry.persistence.PersistenceModule.BeamBulkQueryJpaTm;
 import google.registry.persistence.PersistenceModule.BeamJpaTm;
+import google.registry.persistence.PersistenceModule.BeamReadOnlyReplicaJpaTm;
 import google.registry.persistence.PersistenceModule.TransactionIsolationLevel;
 import google.registry.persistence.transaction.JpaTransactionManager;
 import google.registry.privileges.secretmanager.SecretManagerModule;
@@ -59,6 +60,13 @@ public interface RegistryPipelineComponent {
   @BeamBulkQueryJpaTm
   Lazy<JpaTransactionManager> getBulkQueryJpaTransactionManager();
 
+  /**
+   * A {@link JpaTransactionManager} that uses the Postgres read-only replica if configured (uses
+   * the standard DB otherwise).
+   */
+  @BeamReadOnlyReplicaJpaTm
+  Lazy<JpaTransactionManager> getReadOnlyReplicaJpaTransactionManager();
+
   @Component.Builder
   interface Builder {
 

core/src/main/java/google/registry/beam/common/RegistryPipelineWorkerInitializer.java

@@ -56,6 +56,10 @@ public class RegistryPipelineWorkerInitializer implements JvmInitializer {
       case BULK_QUERY:
         transactionManagerLazy = registryPipelineComponent.getBulkQueryJpaTransactionManager();
         break;
+      case READ_ONLY_REPLICA:
+        transactionManagerLazy =
+            registryPipelineComponent.getReadOnlyReplicaJpaTransactionManager();
+        break;
       case REGULAR:
       default:
         transactionManagerLazy = registryPipelineComponent.getJpaTransactionManager();

core/src/main/java/google/registry/beam/comparedb/DatastoreSnapshots.java

@@ -23,6 +23,7 @@ import com.google.common.collect.ImmutableSet;
 import com.googlecode.objectify.Key;
 import google.registry.backup.VersionedEntity;
 import google.registry.beam.initsql.Transforms;
+import google.registry.model.EppResource;
 import google.registry.model.annotations.DeleteAfterMigration;
 import google.registry.model.billing.BillingEvent;
 import google.registry.model.common.Cursor;
@@ -42,6 +43,7 @@ import google.registry.model.tld.Registry;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
+import javax.annotation.Nullable;
 import org.apache.beam.sdk.Pipeline;
 import org.apache.beam.sdk.transforms.DoFn;
 import org.apache.beam.sdk.transforms.ParDo;
@@ -93,7 +95,8 @@ public final class DatastoreSnapshots {
       String commitLogDir,
       DateTime commitLogFromTime,
       DateTime commitLogToTime,
-      Set<Class<?>> kinds) {
+      Set<Class<?>> kinds,
+      Optional<DateTime> compareStartTime) {
     PCollectionTuple snapshot =
         pipeline.apply(
             "Load Datastore snapshot.",
@@ -112,11 +115,11 @@ public final class DatastoreSnapshots {
         perTypeSnapshots =
             perTypeSnapshots.and(
                 createSqlEntityTupleTag((Class<? extends SqlEntity>) kind),
-                datastoreEntityToPojo(perKindSnapshot, kind.getSimpleName()));
+                datastoreEntityToPojo(perKindSnapshot, kind.getSimpleName(), compareStartTime));
         continue;
       }
       Verify.verify(kind == HistoryEntry.class, "Unexpected Non-SqlEntity class: %s", kind);
-      PCollectionTuple historyEntriesByType = splitHistoryEntry(perKindSnapshot);
+      PCollectionTuple historyEntriesByType = splitHistoryEntry(perKindSnapshot, compareStartTime);
       for (Map.Entry<TupleTag<?>, PCollection<?>> entry :
           historyEntriesByType.getAll().entrySet()) {
         perTypeSnapshots = perTypeSnapshots.and(entry.getKey().getId(), entry.getValue());
@@ -129,7 +132,9 @@ public final class DatastoreSnapshots {
    * Splits a {@link PCollection} of {@link HistoryEntry HistoryEntries} into three collections of
    * its child entities by type.
    */
-  static PCollectionTuple splitHistoryEntry(PCollection<VersionedEntity> historyEntries) {
+  static PCollectionTuple splitHistoryEntry(
+      PCollection<VersionedEntity> historyEntries, Optional<DateTime> compareStartTime) {
+    DateTime nullableStartTime = compareStartTime.orElse(null);
     return historyEntries.apply(
         "Split HistoryEntry by Resource Type",
         ParDo.of(
@@ -138,6 +143,7 @@ public final class DatastoreSnapshots {
                   public void processElement(
                       @Element VersionedEntity historyEntry, MultiOutputReceiver out) {
                     Optional.ofNullable(Transforms.convertVersionedEntityToSqlEntity(historyEntry))
+                        .filter(e -> isEntityIncludedForComparison(e, nullableStartTime))
                         .ifPresent(
                             sqlEntity ->
                                 out.get(createSqlEntityTupleTag(sqlEntity.getClass()))
@@ -155,7 +161,8 @@ public final class DatastoreSnapshots {
    * objects.
    */
   static PCollection<SqlEntity> datastoreEntityToPojo(
-      PCollection<VersionedEntity> entities, String desc) {
+      PCollection<VersionedEntity> entities, String desc, Optional<DateTime> compareStartTime) {
+    DateTime nullableStartTime = compareStartTime.orElse(null);
     return entities.apply(
         "Datastore Entity to Pojo " + desc,
         ParDo.of(
@@ -164,8 +171,23 @@ public final class DatastoreSnapshots {
               public void processElement(
                   @Element VersionedEntity entity, OutputReceiver<SqlEntity> out) {
                 Optional.ofNullable(Transforms.convertVersionedEntityToSqlEntity(entity))
+                    .filter(e -> isEntityIncludedForComparison(e, nullableStartTime))
                     .ifPresent(out::output);
               }
             }));
   }
+
+  static boolean isEntityIncludedForComparison(
+      SqlEntity entity, @Nullable DateTime compareStartTime) {
+    if (compareStartTime == null) {
+      return true;
+    }
+    if (entity instanceof HistoryEntry) {
+      return compareStartTime.isBefore(((HistoryEntry) entity).getModificationTime());
+    }
+    if (entity instanceof EppResource) {
+      return compareStartTime.isBefore(((EppResource) entity).getUpdateTimestamp().getTimestamp());
+    }
+    return true;
+  }
 }

core/src/main/java/google/registry/beam/comparedb/LatestDatastoreSnapshotFinder.java

@@ -53,11 +53,11 @@ public class LatestDatastoreSnapshotFinder {
   }
 
   /**
-   * Finds information of the most recent Datastore snapshot, including the GCS folder of the
-   * exported data files and the start and stop times of the export. The folder of the CommitLogs is
-   * also included in the return.
+   * Finds information of the most recent Datastore snapshot that ends strictly before {@code
+   * exportEndTimeUpperBound}, including the GCS folder of the exported data files and the start and
+   * stop times of the export. The folder of the CommitLogs is also included in the return.
    */
-  public DatastoreSnapshotInfo getSnapshotInfo() {
+  public DatastoreSnapshotInfo getSnapshotInfo(Instant exportEndTimeUpperBound) {
     String bucketName = RegistryConfig.getDatastoreBackupsBucket().substring("gs://".length());
     /**
      * Find the bucket-relative path to the overall metadata file of the last Datastore export.
@@ -65,7 +65,8 @@ public class LatestDatastoreSnapshotFinder {
      * return value is like
      * "2021-11-19T06:00:00_76493/2021-11-19T06:00:00_76493.overall_export_metadata".
      */
-    Optional<String> metaFilePathOptional = findMostRecentExportMetadataFile(bucketName, 2);
+    Optional<String> metaFilePathOptional =
+        findNewestExportMetadataFileBeforeTime(bucketName, exportEndTimeUpperBound, 5);
     if (!metaFilePathOptional.isPresent()) {
       throw new NoSuchElementException("No exports found over the past 2 days.");
     }
@@ -85,8 +86,9 @@ public class LatestDatastoreSnapshotFinder {
   }
 
   /**
-   * Finds the bucket-relative path of the overall export metadata file, in the given bucket,
-   * searching back up to {@code lookBackDays} days, including today.
+   * Finds the latest Datastore export that ends strictly before {@code endTimeUpperBound} and
+   * returns the bucket-relative path of the overall export metadata file, in the given bucket. The
+   * search goes back for up to {@code lookBackDays} days in time, including today.
    *
    * <p>The overall export metadata file is the last file created during a Datastore export. All
    * data has been exported by the creation time of this file. The name of this file, like that of
@@ -95,7 +97,8 @@ public class LatestDatastoreSnapshotFinder {
    * <p>An example return value: {@code
    * 2021-11-19T06:00:00_76493/2021-11-19T06:00:00_76493.overall_export_metadata}.
    */
-  private Optional<String> findMostRecentExportMetadataFile(String bucketName, int lookBackDays) {
+  private Optional<String> findNewestExportMetadataFileBeforeTime(
+      String bucketName, Instant endTimeUpperBound, int lookBackDays) {
     DateTime today = clock.nowUtc();
     for (int day = 0; day < lookBackDays; day++) {
       String dateString = today.minusDays(day).toString("yyyy-MM-dd");
@@ -107,7 +110,11 @@ public class LatestDatastoreSnapshotFinder {
                 .sorted(Comparator.<String>naturalOrder().reversed())
                 .findFirst();
         if (metaFilePath.isPresent()) {
-          return metaFilePath;
+          BlobInfo blobInfo = gcsUtils.getBlobInfo(BlobId.of(bucketName, metaFilePath.get()));
+          Instant exportEndTime = new Instant(blobInfo.getCreateTime());
+          if (exportEndTime.isBefore(endTimeUpperBound)) {
+            return metaFilePath;
+          }
         }
       } catch (IOException ioe) {
         throw new RuntimeException(ioe);
@@ -118,12 +125,12 @@ public class LatestDatastoreSnapshotFinder {
 
   /** Holds information about a Datastore snapshot. */
   @AutoValue
-  abstract static class DatastoreSnapshotInfo {
-    abstract String exportDir();
+  public abstract static class DatastoreSnapshotInfo {
+    public abstract String exportDir();
 
-    abstract String commitLogDir();
+    public abstract String commitLogDir();
 
-    abstract Interval exportInterval();
+    public abstract Interval exportInterval();
 
     static DatastoreSnapshotInfo create(
         String exportDir, String commitLogDir, Interval exportOperationInterval) {

core/src/main/java/google/registry/beam/comparedb/SqlSnapshots.java

@@ -14,15 +14,22 @@
 
 package google.registry.beam.comparedb;
 
+import static com.google.common.base.Preconditions.checkState;
 import static google.registry.beam.comparedb.ValidateSqlUtils.createSqlEntityTupleTag;
 import static google.registry.beam.comparedb.ValidateSqlUtils.getMedianIdForHistoryTable;
 
+import com.google.auto.value.AutoValue;
+import com.google.common.base.Strings;
 import com.google.common.base.Verify;
 import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.ImmutableSetMultimap;
 import com.google.common.collect.Streams;
 import google.registry.beam.common.RegistryJpaIO;
+import google.registry.beam.common.RegistryJpaIO.Read;
+import google.registry.model.EppResource;
+import google.registry.model.UpdateAutoTimestamp;
 import google.registry.model.annotations.DeleteAfterMigration;
 import google.registry.model.billing.BillingEvent;
 import google.registry.model.bulkquery.BulkQueryEntities;
@@ -50,8 +57,10 @@ import google.registry.model.replay.SqlEntity;
 import google.registry.model.reporting.DomainTransactionRecord;
 import google.registry.model.tld.Registry;
 import google.registry.persistence.transaction.CriteriaQueryBuilder;
+import google.registry.util.DateTimeUtils;
 import java.io.Serializable;
 import java.util.Optional;
+import javax.persistence.Entity;
 import org.apache.beam.sdk.Pipeline;
 import org.apache.beam.sdk.transforms.DoFn;
 import org.apache.beam.sdk.transforms.Flatten;
@@ -65,6 +74,7 @@ import org.apache.beam.sdk.values.PCollectionList;
 import org.apache.beam.sdk.values.PCollectionTuple;
 import org.apache.beam.sdk.values.TypeDescriptor;
 import org.apache.beam.sdk.values.TypeDescriptors;
+import org.joda.time.DateTime;
 
 /**
  * Utilities for loading SQL snapshots.
@@ -113,28 +123,48 @@ public final class SqlSnapshots {
   public static PCollectionTuple loadCloudSqlSnapshotByType(
       Pipeline pipeline,
       ImmutableSet<Class<? extends SqlEntity>> sqlEntityTypes,
-      Optional<String> snapshotId) {
+      Optional<String> snapshotId,
+      Optional<DateTime> compareStartTime) {
     PCollectionTuple perTypeSnapshots = PCollectionTuple.empty(pipeline);
     for (Class<? extends SqlEntity> clazz : sqlEntityTypes) {
       if (clazz == DomainBase.class) {
         perTypeSnapshots =
             perTypeSnapshots.and(
                 createSqlEntityTupleTag(DomainBase.class),
-                loadAndAssembleDomainBase(pipeline, snapshotId));
+                loadAndAssembleDomainBase(pipeline, snapshotId, compareStartTime));
         continue;
       }
       if (clazz == DomainHistory.class) {
         perTypeSnapshots =
             perTypeSnapshots.and(
                 createSqlEntityTupleTag(DomainHistory.class),
-                loadAndAssembleDomainHistory(pipeline, snapshotId));
+                loadAndAssembleDomainHistory(pipeline, snapshotId, compareStartTime));
         continue;
       }
       if (clazz == ContactHistory.class) {
         perTypeSnapshots =
             perTypeSnapshots.and(
                 createSqlEntityTupleTag(ContactHistory.class),
-                loadContactHistory(pipeline, snapshotId));
+                loadContactHistory(pipeline, snapshotId, compareStartTime));
+        continue;
+      }
+      if (clazz == HostHistory.class) {
+        perTypeSnapshots =
+            perTypeSnapshots.and(
+                createSqlEntityTupleTag(HostHistory.class),
+                loadHostHistory(
+                    pipeline, snapshotId, compareStartTime.orElse(DateTimeUtils.START_OF_TIME)));
+        continue;
+      }
+      if (EppResource.class.isAssignableFrom(clazz) && compareStartTime.isPresent()) {
+        perTypeSnapshots =
+            perTypeSnapshots.and(
+                createSqlEntityTupleTag(clazz),
+                pipeline.apply(
+                    "SQL Load " + clazz.getSimpleName(),
+                    buildEppResourceQueryWithTimeFilter(
+                            clazz, SqlEntity.class, snapshotId, compareStartTime.get())
+                        .withSnapshot(snapshotId.orElse(null))));
         continue;
       }
       perTypeSnapshots =
@@ -155,20 +185,33 @@ public final class SqlSnapshots {
    * @see BulkQueryEntities
    */
   public static PCollection<SqlEntity> loadAndAssembleDomainBase(
-      Pipeline pipeline, Optional<String> snapshotId) {
+      Pipeline pipeline, Optional<String> snapshotId, Optional<DateTime> compareStartTime) {
     PCollection<KV<String, Serializable>> baseObjects =
-        readAllAndAssignKey(pipeline, DomainBaseLite.class, DomainBaseLite::getRepoId, snapshotId);
+        readAllAndAssignKey(
+            pipeline,
+            DomainBaseLite.class,
+            DomainBaseLite::getRepoId,
+            snapshotId,
+            compareStartTime);
     PCollection<KV<String, Serializable>> gracePeriods =
-        readAllAndAssignKey(pipeline, GracePeriod.class, GracePeriod::getDomainRepoId, snapshotId);
+        readAllAndAssignKey(
+            pipeline,
+            GracePeriod.class,
+            GracePeriod::getDomainRepoId,
+            snapshotId,
+            compareStartTime);
     PCollection<KV<String, Serializable>> delegationSigners =
         readAllAndAssignKey(
             pipeline,
             DelegationSignerData.class,
             DelegationSignerData::getDomainRepoId,
-            snapshotId);
+            snapshotId,
+            compareStartTime);
     PCollection<KV<String, Serializable>> domainHosts =
-        readAllAndAssignKey(pipeline, DomainHost.class, DomainHost::getDomainRepoId, snapshotId);
+        readAllAndAssignKey(
+            pipeline, DomainHost.class, DomainHost::getDomainRepoId, snapshotId, compareStartTime);
 
+    DateTime nullableCompareStartTime = compareStartTime.orElse(null);
     return PCollectionList.of(
             ImmutableList.of(baseObjects, gracePeriods, delegationSigners, domainHosts))
         .apply("SQL Merge DomainBase parts", Flatten.pCollections())
@@ -184,6 +227,14 @@ public final class SqlSnapshots {
                     TypedClassifier partsByType = new TypedClassifier(kv.getValue());
                     ImmutableSet<DomainBaseLite> baseObjects =
                         partsByType.getAllOf(DomainBaseLite.class);
+                    if (nullableCompareStartTime != null) {
+                      Verify.verify(
+                          baseObjects.size() <= 1,
+                          "Found duplicate DomainBaseLite object per repoId: " + kv.getKey());
+                      if (baseObjects.isEmpty()) {
+                        return;
+                      }
+                    }
                     Verify.verify(
                         baseObjects.size() == 1,
                         "Expecting one DomainBaseLite object per repoId: " + kv.getKey());
@@ -205,16 +256,16 @@ public final class SqlSnapshots {
    * <p>This method uses two queries to load data in parallel. This is a performance optimization
    * specifically for the production database.
    */
-  static PCollection<SqlEntity> loadContactHistory(Pipeline pipeline, Optional<String> snapshotId) {
-    long medianId =
-        getMedianIdForHistoryTable("ContactHistory")
-            .orElseThrow(
-                () -> new IllegalStateException("Not a valid database: no ContactHistory."));
+  static PCollection<SqlEntity> loadContactHistory(
+      Pipeline pipeline, Optional<String> snapshotId, Optional<DateTime> compareStartTime) {
+    PartitionedQuery partitionedQuery =
+        buildPartitonedHistoryQuery(ContactHistory.class, compareStartTime);
     PCollection<SqlEntity> part1 =
         pipeline.apply(
             "SQL Load ContactHistory first half",
             RegistryJpaIO.read(
-                    String.format("select c from ContactHistory c where id <= %s", medianId),
+                    partitionedQuery.firstHalfQuery(),
+                    partitionedQuery.parameters(),
                     false,
                     SqlEntity.class::cast)
                 .withSnapshot(snapshotId.orElse(null)));
@@ -222,7 +273,8 @@ public final class SqlSnapshots {
         pipeline.apply(
             "SQL Load ContactHistory second half",
             RegistryJpaIO.read(
-                    String.format("select c from ContactHistory c where id > %s", medianId),
+                    partitionedQuery.secondHalfQuery(),
+                    partitionedQuery.parameters(),
                     false,
                     SqlEntity.class::cast)
                 .withSnapshot(snapshotId.orElse(null)));
@@ -231,6 +283,19 @@ public final class SqlSnapshots {
         .apply("Combine ContactHistory parts", Flatten.pCollections());
   }
 
+  /** Loads all {@link HostHistory} entities from the database. */
+  static PCollection<SqlEntity> loadHostHistory(
+      Pipeline pipeline, Optional<String> snapshotId, DateTime compareStartTime) {
+    return pipeline.apply(
+        "SQL Load HostHistory",
+        RegistryJpaIO.read(
+                "select c from HostHistory c where :compareStartTime  < modificationTime",
+                ImmutableMap.of("compareStartTime", compareStartTime),
+                false,
+                SqlEntity.class::cast)
+            .withSnapshot(snapshotId.orElse(null)));
+  }
+
   /**
    * Bulk-loads all parts of {@link DomainHistory} and assembles them in the pipeline.
    *
@@ -240,16 +305,15 @@ public final class SqlSnapshots {
    * @see BulkQueryEntities
    */
   static PCollection<SqlEntity> loadAndAssembleDomainHistory(
-      Pipeline pipeline, Optional<String> snapshotId) {
-    long medianId =
-        getMedianIdForHistoryTable("DomainHistory")
-            .orElseThrow(
-                () -> new IllegalStateException("Not a valid database: no DomainHistory."));
+      Pipeline pipeline, Optional<String> snapshotId, Optional<DateTime> compareStartTime) {
+    PartitionedQuery partitionedQuery =
+        buildPartitonedHistoryQuery(DomainHistoryLite.class, compareStartTime);
     PCollection<KV<String, Serializable>> baseObjectsPart1 =
         queryAndAssignKey(
             pipeline,
             "first half",
-            String.format("select c from DomainHistory c where id <= %s", medianId),
+            partitionedQuery.firstHalfQuery(),
+            partitionedQuery.parameters(),
             DomainHistoryLite.class,
             compose(DomainHistoryLite::getDomainHistoryId, DomainHistoryId::toString),
             snapshotId);
@@ -257,7 +321,8 @@ public final class SqlSnapshots {
         queryAndAssignKey(
             pipeline,
             "second half",
-            String.format("select c from DomainHistory c where id > %s", medianId),
+            partitionedQuery.secondHalfQuery(),
+            partitionedQuery.parameters(),
             DomainHistoryLite.class,
             compose(DomainHistoryLite::getDomainHistoryId, DomainHistoryId::toString),
             snapshotId);
@@ -266,26 +331,31 @@ public final class SqlSnapshots {
             pipeline,
             GracePeriodHistory.class,
             compose(GracePeriodHistory::getDomainHistoryId, DomainHistoryId::toString),
-            snapshotId);
+            snapshotId,
+            compareStartTime);
     PCollection<KV<String, Serializable>> delegationSigners =
         readAllAndAssignKey(
             pipeline,
             DomainDsDataHistory.class,
             compose(DomainDsDataHistory::getDomainHistoryId, DomainHistoryId::toString),
-            snapshotId);
+            snapshotId,
+            compareStartTime);
     PCollection<KV<String, Serializable>> domainHosts =
         readAllAndAssignKey(
             pipeline,
             DomainHistoryHost.class,
             compose(DomainHistoryHost::getDomainHistoryId, DomainHistoryId::toString),
-            snapshotId);
+            snapshotId,
+            compareStartTime);
     PCollection<KV<String, Serializable>> transactionRecords =
         readAllAndAssignKey(
             pipeline,
             DomainTransactionRecord.class,
             compose(DomainTransactionRecord::getDomainHistoryId, DomainHistoryId::toString),
-            snapshotId);
+            snapshotId,
+            compareStartTime);
 
+    DateTime nullableCompareStartTime = compareStartTime.orElse(null);
     return PCollectionList.of(
             ImmutableList.of(
                 baseObjectsPart1,
@@ -307,6 +377,15 @@ public final class SqlSnapshots {
                     TypedClassifier partsByType = new TypedClassifier(kv.getValue());
                     ImmutableSet<DomainHistoryLite> baseObjects =
                         partsByType.getAllOf(DomainHistoryLite.class);
+                    if (nullableCompareStartTime != null) {
+                      Verify.verify(
+                          baseObjects.size() <= 1,
+                          "Found duplicate DomainHistoryLite object per domainHistoryId: "
+                              + kv.getKey());
+                      if (baseObjects.isEmpty()) {
+                        return;
+                      }
+                    }
                     Verify.verify(
                         baseObjects.size() == 1,
                         "Expecting one DomainHistoryLite object per domainHistoryId: "
@@ -328,12 +407,19 @@ public final class SqlSnapshots {
       Pipeline pipeline,
       Class<R> type,
       SerializableFunction<R, String> keyFunction,
-      Optional<String> snapshotId) {
+      Optional<String> snapshotId,
+      Optional<DateTime> compareStartTime) {
+    Read<R, R> queryObject;
+    if (compareStartTime.isPresent() && EppResource.class.isAssignableFrom(type)) {
+      queryObject =
+          buildEppResourceQueryWithTimeFilter(type, type, snapshotId, compareStartTime.get());
+    } else {
+      queryObject =
+          RegistryJpaIO.read(() -> CriteriaQueryBuilder.create(type).build())
+              .withSnapshot(snapshotId.orElse(null));
+    }
     return pipeline
-        .apply(
-            "SQL Load " + type.getSimpleName(),
-            RegistryJpaIO.read(() -> CriteriaQueryBuilder.create(type).build())
-                .withSnapshot(snapshotId.orElse(null)))
+        .apply("SQL Load " + type.getSimpleName(), queryObject)
         .apply(
             "Assign Key to " + type.getSimpleName(),
             MapElements.into(
@@ -346,13 +432,15 @@ public final class SqlSnapshots {
       Pipeline pipeline,
       String diffrentiator,
       String jplQuery,
+      ImmutableMap<String, Object> queryParameters,
       Class<R> type,
       SerializableFunction<R, String> keyFunction,
       Optional<String> snapshotId) {
     return pipeline
         .apply(
             "SQL Load " + type.getSimpleName() + " " + diffrentiator,
-            RegistryJpaIO.read(jplQuery, false, type::cast).withSnapshot(snapshotId.orElse(null)))
+            RegistryJpaIO.read(jplQuery, queryParameters, false, type::cast)
+                .withSnapshot(snapshotId.orElse(null)))
         .apply(
             "Assign Key to " + type.getSimpleName() + " " + diffrentiator,
             MapElements.into(
@@ -367,6 +455,71 @@ public final class SqlSnapshots {
     return r -> f2.apply(f1.apply(r));
   }
 
+  static <R, T> Read<R, T> buildEppResourceQueryWithTimeFilter(
+      Class<R> entityType,
+      Class<T> castOutputAsType,
+      Optional<String> snapshotId,
+      DateTime compareStartTime) {
+    String tableName = getJpaEntityName(entityType);
+    String jpql =
+        String.format("select c from %s c where :compareStartTime < updateTimestamp", tableName);
+    return RegistryJpaIO.read(
+            jpql,
+            ImmutableMap.of("compareStartTime", UpdateAutoTimestamp.create(compareStartTime)),
+            false,
+            (R x) -> castOutputAsType.cast(x))
+        .withSnapshot(snapshotId.orElse(null));
+  }
+
+  static PartitionedQuery buildPartitonedHistoryQuery(
+      Class<?> entityType, Optional<DateTime> compareStartTime) {
+    String tableName = getJpaEntityName(entityType);
+    Verify.verify(
+        !Strings.isNullOrEmpty(tableName), "Invalid entity type %s", entityType.getSimpleName());
+    long medianId =
+        getMedianIdForHistoryTable(tableName)
+            .orElseThrow(() -> new IllegalStateException("Not a valid database: no " + tableName));
+    String firstHalfQuery = String.format("select c from %s c where id <= :historyId", tableName);
+    String secondHalfQuery = String.format("select c from %s c where id > :historyId", tableName);
+    if (compareStartTime.isPresent()) {
+      String timeFilter = "  and :compareStartTime  < modificationTime";
+      firstHalfQuery += timeFilter;
+      secondHalfQuery += timeFilter;
+      return PartitionedQuery.createPartitionedQuery(
+          firstHalfQuery,
+          secondHalfQuery,
+          ImmutableMap.of("historyId", medianId, "compareStartTime", compareStartTime.get()));
+    } else {
+      return PartitionedQuery.createPartitionedQuery(
+          firstHalfQuery, secondHalfQuery, ImmutableMap.of("historyId", medianId));
+    }
+  }
+
+  private static String getJpaEntityName(Class entityType) {
+    Entity entityAnnotation = (Entity) entityType.getAnnotation(Entity.class);
+    checkState(
+        entityAnnotation != null, "Unexpected non-entity type %s", entityType.getSimpleName());
+    return Strings.isNullOrEmpty(entityAnnotation.name())
+        ? entityType.getSimpleName()
+        : entityAnnotation.name();
+  }
+
+  /** Contains two queries that partition the target table in two. */
+  @AutoValue
+  abstract static class PartitionedQuery {
+    abstract String firstHalfQuery();
+
+    abstract String secondHalfQuery();
+
+    abstract ImmutableMap<String, Object> parameters();
+
+    public static PartitionedQuery createPartitionedQuery(
+        String firstHalfQuery, String secondHalfQuery, ImmutableMap<String, Object> parameters) {
+      return new AutoValue_SqlSnapshots_PartitionedQuery(
+          firstHalfQuery, secondHalfQuery, parameters);
+    }
+  }
+
   /** Container that receives mixed-typed data and groups them by {@link Class}. */
   static class TypedClassifier {
     private final ImmutableSetMultimap<Class<?>, Object> classifiedEntities;

core/src/main/java/google/registry/beam/comparedb/ValidateDatastorePipeline.java

@@ -0,0 +1,83 @@
+// 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.beam.comparedb;
+
+import google.registry.beam.common.RegistryPipelineOptions;
+import google.registry.beam.common.RegistryPipelineWorkerInitializer;
+import google.registry.beam.comparedb.LatestDatastoreSnapshotFinder.DatastoreSnapshotInfo;
+import google.registry.model.annotations.DeleteAfterMigration;
+import google.registry.persistence.PersistenceModule.JpaTransactionManagerType;
+import google.registry.persistence.PersistenceModule.TransactionIsolationLevel;
+import java.util.Optional;
+import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.options.PipelineOptionsFactory;
+import org.joda.time.DateTime;
+
+/**
+ * Validates the asynchronous data replication process from Cloud SQL (primary) to Datastore
+ * (secondary).
+ *
+ * <p>This pipeline simply compares the snapshots provided by an invoker, which is responsible for
+ * obtaining two consistent snapshots for the same point of time.
+ */
+// TODO(weiminyu): Implement the invoker action in a followup PR.
+@DeleteAfterMigration
+public class ValidateDatastorePipeline {
+
+  private final ValidateDatastorePipelineOptions options;
+  private final LatestDatastoreSnapshotFinder datastoreSnapshotFinder;
+
+  public ValidateDatastorePipeline(
+      ValidateDatastorePipelineOptions options,
+      LatestDatastoreSnapshotFinder datastoreSnapshotFinder) {
+    this.options = options;
+    this.datastoreSnapshotFinder = datastoreSnapshotFinder;
+  }
+
+  void run(Pipeline pipeline) {
+    DateTime latestCommitLogTime = DateTime.parse(options.getLatestCommitLogTimestamp());
+    DatastoreSnapshotInfo mostRecentExport =
+        datastoreSnapshotFinder.getSnapshotInfo(latestCommitLogTime.toInstant());
+
+    ValidateSqlPipeline.setupPipeline(
+        pipeline,
+        Optional.ofNullable(options.getSqlSnapshotId()),
+        mostRecentExport,
+        latestCommitLogTime,
+        Optional.ofNullable(options.getComparisonStartTimestamp()).map(DateTime::parse));
+
+    pipeline.run();
+  }
+
+  public static void main(String[] args) {
+    ValidateDatastorePipelineOptions options =
+        PipelineOptionsFactory.fromArgs(args)
+            .withValidation()
+            .as(ValidateDatastorePipelineOptions.class);
+    RegistryPipelineOptions.validateRegistryPipelineOptions(options);
+
+    // Defensively set important options.
+    options.setIsolationOverride(TransactionIsolationLevel.TRANSACTION_REPEATABLE_READ);
+    options.setJpaTransactionManagerType(JpaTransactionManagerType.BULK_QUERY);
+
+    // Reuse Dataflow worker initialization code to set up JPA in the pipeline harness.
+    new RegistryPipelineWorkerInitializer().beforeProcessing(options);
+
+    LatestDatastoreSnapshotFinder datastoreSnapshotFinder =
+        DaggerLatestDatastoreSnapshotFinder_LatestDatastoreSnapshotFinderFinderComponent.create()
+            .datastoreSnapshotInfoFinder();
+    new ValidateDatastorePipeline(options, datastoreSnapshotFinder).run(Pipeline.create(options));
+  }
+}

core/src/main/java/google/registry/beam/comparedb/ValidateDatastorePipelineOptions.java

@@ -0,0 +1,39 @@
+// 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.beam.comparedb;
+
+import google.registry.model.annotations.DeleteAfterMigration;
+import javax.annotation.Nullable;
+import org.apache.beam.sdk.options.Description;
+import org.apache.beam.sdk.options.Validation;
+
+/** BEAM pipeline options for {@link ValidateDatastorePipelineOptions}. */
+@DeleteAfterMigration
+public interface ValidateDatastorePipelineOptions extends ValidateSqlPipelineOptions {
+
+  @Description(
+      "The id of the SQL snapshot to be compared with Datastore. "
+          + "If null, the current state of the SQL database is used.")
+  @Nullable
+  String getSqlSnapshotId();
+
+  void setSqlSnapshotId(String snapshotId);
+
+  @Description("The latest CommitLogs to load, in ISO8601 format.")
+  @Validation.Required
+  String getLatestCommitLogTimestamp();
+
+  void setLatestCommitLogTimestamp(String commitLogEndTimestamp);
+}

core/src/main/java/google/registry/beam/comparedb/ValidateSqlPipeline.java

@@ -26,6 +26,9 @@ import google.registry.beam.common.RegistryPipelineWorkerInitializer;
 import google.registry.beam.comparedb.LatestDatastoreSnapshotFinder.DatastoreSnapshotInfo;
 import google.registry.beam.comparedb.ValidateSqlUtils.CompareSqlEntity;
 import google.registry.model.annotations.DeleteAfterMigration;
+import google.registry.model.common.DatabaseMigrationStateSchedule;
+import google.registry.model.common.DatabaseMigrationStateSchedule.MigrationState;
+import google.registry.model.common.DatabaseMigrationStateSchedule.ReplayDirection;
 import google.registry.model.domain.DomainBase;
 import google.registry.model.domain.DomainHistory;
 import google.registry.model.replay.SqlEntity;
@@ -35,6 +38,7 @@ import google.registry.persistence.PersistenceModule.JpaTransactionManagerType;
 import google.registry.persistence.PersistenceModule.TransactionIsolationLevel;
 import google.registry.persistence.transaction.TransactionManagerFactory;
 import google.registry.util.RequestStatusChecker;
+import google.registry.util.SystemClock;
 import java.io.Serializable;
 import java.util.Optional;
 import org.apache.beam.sdk.Pipeline;
@@ -76,21 +80,16 @@ public class ValidateSqlPipeline {
       java.time.Duration.ofSeconds(30);
 
   private final ValidateSqlPipelineOptions options;
-  private final DatastoreSnapshotInfo mostRecentExport;
+  private final LatestDatastoreSnapshotFinder datastoreSnapshotFinder;
 
   public ValidateSqlPipeline(
-      ValidateSqlPipelineOptions options, DatastoreSnapshotInfo mostRecentExport) {
+      ValidateSqlPipelineOptions options, LatestDatastoreSnapshotFinder datastoreSnapshotFinder) {
     this.options = options;
-    this.mostRecentExport = mostRecentExport;
-  }
-
-  void run() {
-    run(Pipeline.create(options));
+    this.datastoreSnapshotFinder = datastoreSnapshotFinder;
   }
 
   @VisibleForTesting
   void run(Pipeline pipeline) {
-    // TODO(weiminyu): ensure migration stage is DATASTORE_PRIMARY or DATASTORE_PRIMARY_READ_ONLY
     Optional<Lock> lock = acquireCommitLogReplayLock();
     if (lock.isPresent()) {
       logger.atInfo().log("Acquired CommitLog Replay lock.");
@@ -101,6 +100,8 @@ public class ValidateSqlPipeline {
     try {
       DateTime latestCommitLogTime =
           TransactionManagerFactory.jpaTm().transact(() -> SqlReplayCheckpoint.get());
+      DatastoreSnapshotInfo mostRecentExport =
+          datastoreSnapshotFinder.getSnapshotInfo(latestCommitLogTime.toInstant());
       Preconditions.checkState(
           latestCommitLogTime.isAfter(mostRecentExport.exportInterval().getEnd()),
           "Cannot recreate Datastore snapshot since target time is in the middle of an export.");
@@ -109,7 +110,16 @@ public class ValidateSqlPipeline {
         lock.ifPresent(Lock::releaseSql);
         lock = Optional.empty();
 
-        setupPipeline(pipeline, Optional.of(databaseSnapshot.getSnapshotId()), latestCommitLogTime);
+        logger.atInfo().log(
+            "Starting comparison with export at %s and latestCommitLogTime at %s",
+            mostRecentExport.exportDir(), latestCommitLogTime);
+
+        setupPipeline(
+            pipeline,
+            Optional.of(databaseSnapshot.getSnapshotId()),
+            mostRecentExport,
+            latestCommitLogTime,
+            Optional.ofNullable(options.getComparisonStartTimestamp()).map(DateTime::parse));
         State state = pipeline.run().waitUntilFinish();
         if (!State.DONE.equals(state)) {
           throw new IllegalStateException("Unexpected pipeline state: " + state);
@@ -120,8 +130,12 @@ public class ValidateSqlPipeline {
     }
   }
 
-  void setupPipeline(
-      Pipeline pipeline, Optional<String> sqlSnapshotId, DateTime latestCommitLogTime) {
+  static void setupPipeline(
+      Pipeline pipeline,
+      Optional<String> sqlSnapshotId,
+      DatastoreSnapshotInfo mostRecentExport,
+      DateTime latestCommitLogTime,
+      Optional<DateTime> compareStartTime) {
     pipeline
         .getCoderRegistry()
         .registerCoderForClass(SqlEntity.class, SerializableCoder.of(Serializable.class));
@@ -135,11 +149,12 @@ public class ValidateSqlPipeline {
             // Increase by 1ms since we want to include commitLogs latestCommitLogTime but
             // this parameter is exclusive.
             latestCommitLogTime.plusMillis(1),
-            DatastoreSnapshots.ALL_DATASTORE_KINDS);
+            DatastoreSnapshots.ALL_DATASTORE_KINDS,
+            compareStartTime);
 
     PCollectionTuple cloudSqlSnapshot =
         SqlSnapshots.loadCloudSqlSnapshotByType(
-            pipeline, SqlSnapshots.ALL_SQL_ENTITIES, sqlSnapshotId);
+            pipeline, SqlSnapshots.ALL_SQL_ENTITIES, sqlSnapshotId, compareStartTime);
 
     verify(
         datastoreSnapshot.getAll().keySet().equals(cloudSqlSnapshot.getAll().keySet()),
@@ -212,6 +227,10 @@ public class ValidateSqlPipeline {
         return "ValidateSqlPipeline";
       }
 
+      LatestDatastoreSnapshotFinder datastoreSnapshotFinder =
+          DaggerLatestDatastoreSnapshotFinder_LatestDatastoreSnapshotFinderFinderComponent.create()
+              .datastoreSnapshotInfoFinder();
+
       @Override
       public boolean isRunning(String requestLogId) {
         return true;
@@ -230,11 +249,16 @@ public class ValidateSqlPipeline {
     // Reuse Dataflow worker initialization code to set up JPA in the pipeline harness.
     new RegistryPipelineWorkerInitializer().beforeProcessing(options);
 
-    DatastoreSnapshotInfo mostRecentExport =
-        DaggerLatestDatastoreSnapshotFinder_LatestDatastoreSnapshotFinderFinderComponent.create()
-            .datastoreSnapshotInfoFinder()
-            .getSnapshotInfo();
+    MigrationState state =
+        DatabaseMigrationStateSchedule.getValueAtTime(new SystemClock().nowUtc());
+    if (!state.getReplayDirection().equals(ReplayDirection.DATASTORE_TO_SQL)) {
+      throw new IllegalStateException("This pipeline is not designed for migration phase " + state);
+    }
 
-    new ValidateSqlPipeline(options, mostRecentExport).run(Pipeline.create(options));
+    LatestDatastoreSnapshotFinder datastoreSnapshotFinder =
+        DaggerLatestDatastoreSnapshotFinder_LatestDatastoreSnapshotFinderFinderComponent.create()
+            .datastoreSnapshotInfoFinder();
+
+    new ValidateSqlPipeline(options, datastoreSnapshotFinder).run(Pipeline.create(options));
   }
 }

core/src/main/java/google/registry/beam/comparedb/ValidateSqlPipelineOptions.java

@@ -16,7 +16,19 @@ package google.registry.beam.comparedb;
 
 import google.registry.beam.common.RegistryPipelineOptions;
 import google.registry.model.annotations.DeleteAfterMigration;
+import javax.annotation.Nullable;
+import org.apache.beam.sdk.options.Description;
 
 /** BEAM pipeline options for {@link ValidateSqlPipeline}. */
 @DeleteAfterMigration
-public interface ValidateSqlPipelineOptions extends RegistryPipelineOptions {}
+public interface ValidateSqlPipelineOptions extends RegistryPipelineOptions {
+
+  @Description(
+      "For history entries and EPP resources, only those modified strictly after this time are "
+          + "included in comparison. Value is in ISO8601 format. "
+          + "Other entity types are not affected.")
+  @Nullable
+  String getComparisonStartTimestamp();
+
+  void setComparisonStartTimestamp(String comparisonStartTimestamp);
+}

core/src/main/java/google/registry/beam/comparedb/ValidateSqlUtils.java

@@ -104,6 +104,7 @@ final class ValidateSqlUtils {
     private final HashMap<String, Counter> missingCounters = new HashMap<>();
     private final HashMap<String, Counter> unequalCounters = new HashMap<>();
     private final HashMap<String, Counter> badEntityCounters = new HashMap<>();
+    private final HashMap<String, Counter> duplicateEntityCounters = new HashMap<>();
 
     private volatile boolean logPrinted = false;
 
@@ -120,6 +121,8 @@ final class ValidateSqlUtils {
           counterKey, Metrics.counter("CompareDB", "Missing In One DB: " + counterKey));
       unequalCounters.put(counterKey, Metrics.counter("CompareDB", "Not Equal:" + counterKey));
       badEntityCounters.put(counterKey, Metrics.counter("CompareDB", "Bad Entities:" + counterKey));
+      duplicateEntityCounters.put(
+          counterKey, Metrics.counter("CompareDB", "Duplicate Entities:" + counterKey));
     }
 
     /**
@@ -158,12 +161,18 @@ final class ValidateSqlUtils {
       ImmutableList<SqlEntity> entities = ImmutableList.copyOf(kv.getValue());
 
       verify(!entities.isEmpty(), "Can't happen: no value for key %s.", kv.getKey());
-      verify(entities.size() <= 2, "Unexpected duplicates for key %s", kv.getKey());
 
       String counterKey = getCounterKey(entities.get(0).getClass());
       ensureCounterExists(counterKey);
       totalCounters.get(counterKey).inc();
 
+      if (entities.size() > 2) {
+        // Duplicates may happen with Cursors if imported across projects. Its key in Datastore, the
+        // id field, encodes the project name and is not fixed by the importing job.
+        duplicateEntityCounters.get(counterKey).inc();
+        return;
+      }
+
       if (entities.size() == 1) {
         if (isSpecialCaseProberEntity(entities.get(0))) {
           return;

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

@@ -248,6 +248,9 @@ public class RdeIO {
       // Now that we're done, output roll the cursor forward.
       if (key.manual()) {
         logger.atInfo().log("Manual operation; not advancing cursor or enqueuing upload task.");
+        // Temporary measure to run RDE in beam in parallel with the daily MapReduce based RDE runs.
+      } else if (tm().isOfy()) {
+        logger.atInfo().log("Ofy is primary TM; not advancing cursor or enqueuing upload task.");
       } else {
         outputReceiver.output(KV.of(key, revision));
       }
@@ -294,6 +297,10 @@ public class RdeIO {
                 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);
+                // 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.
                 if (key.mode() == RdeMode.FULL) {
                   cloudTasksUtils.enqueue(
                       RDE_UPLOAD_QUEUE,

core/src/main/java/google/registry/config/RegistryConfig.java

@@ -33,6 +33,7 @@ import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.ImmutableSortedMap;
 import dagger.Module;
 import dagger.Provides;
+import google.registry.persistence.transaction.JpaTransactionManager;
 import google.registry.util.TaskQueueUtils;
 import google.registry.util.YamlUtils;
 import java.lang.annotation.Documented;
@@ -392,19 +393,26 @@ public final class RegistryConfig {
 
     @Provides
     @Config("cloudSqlJdbcUrl")
-    public static String providesCloudSqlJdbcUrl(RegistryConfigSettings config) {
+    public static String provideCloudSqlJdbcUrl(RegistryConfigSettings config) {
       return config.cloudSql.jdbcUrl;
     }
 
     @Provides
     @Config("cloudSqlInstanceConnectionName")
-    public static String providesCloudSqlInstanceConnectionName(RegistryConfigSettings config) {
+    public static String provideCloudSqlInstanceConnectionName(RegistryConfigSettings config) {
       return config.cloudSql.instanceConnectionName;
     }
 
+    @Provides
+    @Config("cloudSqlReplicaInstanceConnectionName")
+    public static Optional<String> provideCloudSqlReplicaInstanceConnectionName(
+        RegistryConfigSettings config) {
+      return Optional.ofNullable(config.cloudSql.replicaInstanceConnectionName);
+    }
+
     @Provides
     @Config("cloudSqlDbInstanceName")
-    public static String providesCloudSqlDbInstance(RegistryConfigSettings config) {
+    public static String provideCloudSqlDbInstance(RegistryConfigSettings config) {
       // Format of instanceConnectionName: project-id:region:instance-name
       int lastColonIndex = config.cloudSql.instanceConnectionName.lastIndexOf(':');
       return config.cloudSql.instanceConnectionName.substring(lastColonIndex + 1);
@@ -1524,6 +1532,31 @@ public final class RegistryConfig {
     return CONFIG_SETTINGS.get().hibernate.hikariIdleTimeout;
   }
 
+  /**
+   * JDBC-specific: driver default batch size is 0, which means that every INSERT statement will be
+   * sent to the database individually. Batching allows us to group together multiple inserts into
+   * one single INSERT statement which can dramatically increase speed in situations with many
+   * inserts.
+   *
+   * <p>Hibernate docs, i.e.
+   * https://docs.jboss.org/hibernate/orm/5.6/userguide/html_single/Hibernate_User_Guide.html,
+   * recommend between 10 and 50.
+   */
+  public static String getHibernateJdbcBatchSize() {
+    return CONFIG_SETTINGS.get().hibernate.jdbcBatchSize;
+  }
+
+  /**
+   * Returns the JDBC fetch size.
+   *
+   * <p>Postgresql-specific: driver default fetch size is 0, which disables streaming result sets.
+   * Here we set a small default geared toward Nomulus server transactions. Large queries can
+   * override the defaults using {@link JpaTransactionManager#setQueryFetchSize}.
+   */
+  public static String getHibernateJdbcFetchSize() {
+    return CONFIG_SETTINGS.get().hibernate.jdbcFetchSize;
+  }
+
   /** Returns the roid suffix to be used for the roids of all contacts and hosts. */
   public static String getContactAndHostRoidSuffix() {
     return CONFIG_SETTINGS.get().registryPolicy.contactAndHostRoidSuffix;

core/src/main/java/google/registry/config/RegistryConfigSettings.java

@@ -120,6 +120,8 @@ public class RegistryConfigSettings {
     public String hikariMinimumIdle;
     public String hikariMaximumPoolSize;
     public String hikariIdleTimeout;
+    public String jdbcBatchSize;
+    public String jdbcFetchSize;
   }
 
   /** Configuration for Cloud SQL. */
@@ -128,6 +130,7 @@ public class RegistryConfigSettings {
     // TODO(05012021): remove username field after it is removed from all yaml files.
     public String username;
     public String instanceConnectionName;
+    public String replicaInstanceConnectionName;
   }
 
   /** Configuration for Apache Beam (Cloud Dataflow). */

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

@@ -221,6 +221,17 @@ hibernate:
   hikariMinimumIdle: 1
   hikariMaximumPoolSize: 10
   hikariIdleTimeout: 300000
+  # The batch size is basically the number of insertions / updates in a single
+  # transaction that will be batched together into one INSERT/UPDATE statement.
+  # A larger batch size is useful when inserting or updating many entities in a
+  # single transaction. Hibernate docs
+  # (https://docs.jboss.org/hibernate/orm/5.6/userguide/html_single/Hibernate_User_Guide.html)
+  # recommend between 10 and 50.
+  jdbcBatchSize: 50
+  # The fetch size is the number of entities retrieved at a time from the
+  # database cursor. Here we set a small default geared toward Nomulus server
+  # transactions. Large queries can override the defaults on a per-query basis.
+  jdbcFetchSize: 20
 
 cloudSql:
   # jdbc url for the Cloud SQL database.
@@ -231,6 +242,10 @@ cloudSql:
   jdbcUrl: jdbc:postgresql://localhost
   # This name is used by Cloud SQL when connecting to the database.
   instanceConnectionName: project-id:region:instance-id
+  # If non-null, we will use this instance for certain read-only actions or
+  # pipelines, e.g. RDE, in order to offload some work from the primary
+  # instance. Expect any write actions on this instance to fail.
+  replicaInstanceConnectionName: null
 
 cloudDns:
   # Set both properties to null in Production.

core/src/main/java/google/registry/env/common/backend/WEB-INF/web.xml

@@ -422,6 +422,12 @@ have been in the database for a certain period of time. -->
     <url-pattern>/_dr/task/createSyntheticHistoryEntries</url-pattern>
   </servlet-mapping>
 
+  <!-- Action to sync Datastore to a snapshot of the primary SQL database. -->
+  <servlet-mapping>
+    <servlet-name>backend-servlet</servlet-name>
+    <url-pattern>/_dr/task/syncDatastoreToSqlSnapshot</url-pattern>
+  </servlet-mapping>
+
   <!-- Security config -->
   <security-constraint>
     <web-resource-collection>

core/src/main/java/google/registry/env/production/default/WEB-INF/cron.xml

@@ -36,6 +36,19 @@
     <target>backend</target>
   </cron>
 
+  <cron>
+    <url>/_dr/task/rdeStaging?beam=true</url>
+    <description>
+      This job generates a full RDE escrow deposit as a single gigantic XML
+      document using the Beam pipeline regardless of the current TM
+      configuration and streams it to cloud storage. It does not trigger the
+      subsequent upload tasks and is meant to run parallel with the main cron
+      job in order to compare the results from both runs.
+    </description>
+    <schedule>every 8 hours from 00:07 to 20:00</schedule>
+    <target>backend</target>
+  </cron>
+
   <cron>
     <url><![CDATA[/_dr/cron/fanout?queue=rde-upload&endpoint=/_dr/task/rdeUpload&forEachRealTld]]></url>
     <description>
@@ -349,6 +362,15 @@
     <target>backend</target>
   </cron>
 
+  <cron>
+    <url><![CDATA[/_dr/cron/replicateToDatastore]]></url>
+    <description>
+      Replays recent transactions from SQL to the Datastore secondary backend.
+    </description>
+    <schedule>every 3 minutes</schedule>
+    <target>backend</target>
+  </cron>
+
   <cron>
     <url><![CDATA[/_dr/task/wipeOutContactHistoryPii]]></url>
     <description>

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

@@ -26,7 +26,6 @@ import com.google.common.base.Joiner;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableSortedSet;
 import com.google.common.collect.Iterables;
-import com.google.common.collect.Streams;
 import com.google.common.flogger.FluentLogger;
 import com.google.common.net.MediaType;
 import google.registry.config.RegistryConfig.Config;
@@ -143,7 +142,7 @@ public class ExportPremiumTermsAction implements Runnable {
         PremiumListDao.getLatestRevision(premiumListName).isPresent(),
         "Could not load premium list for " + tld);
     SortedSet<String> premiumTerms =
-        Streams.stream(PremiumListDao.loadAllPremiumEntries(premiumListName))
+        PremiumListDao.loadAllPremiumEntries(premiumListName).stream()
             .map(PremiumEntry::toString)
             .collect(ImmutableSortedSet.toImmutableSortedSet(String::compareTo));
 

core/src/main/java/google/registry/export/sheet/SyncRegistrarsSheetAction.java

@@ -14,8 +14,6 @@
 
 package google.registry.export.sheet;
 
-import static com.google.appengine.api.taskqueue.QueueFactory.getQueue;
-import static com.google.appengine.api.taskqueue.TaskOptions.Builder.withUrl;
 import static com.google.common.net.MediaType.PLAIN_TEXT_UTF_8;
 import static google.registry.request.Action.Method.POST;
 import static javax.servlet.http.HttpServletResponse.SC_BAD_REQUEST;
@@ -23,7 +21,6 @@ import static javax.servlet.http.HttpServletResponse.SC_INTERNAL_SERVER_ERROR;
 import static javax.servlet.http.HttpServletResponse.SC_NO_CONTENT;
 import static javax.servlet.http.HttpServletResponse.SC_OK;
 
-import com.google.appengine.api.taskqueue.TaskOptions.Method;
 import com.google.common.flogger.FluentLogger;
 import google.registry.config.RegistryConfig.Config;
 import google.registry.request.Action;
@@ -100,7 +97,7 @@ public class SyncRegistrarsSheetAction implements Runnable {
   }
 
   public static final String PATH = "/_dr/task/syncRegistrarsSheet";
-  private static final String QUEUE = "sheet";
+  public static final String QUEUE = "sheet";
   private static final String LOCK_NAME = "Synchronize registrars sheet";
   private static final FluentLogger logger = FluentLogger.forEnclosingClass();
 
@@ -144,11 +141,4 @@ public class SyncRegistrarsSheetAction implements Runnable {
       Result.LOCKED.send(response, null);
     }
   }
-
-  /**
-   * Enqueues a sync registrar sheet task targeting the App Engine service specified by hostname.
-   */
-  public static void enqueueRegistrarSheetSync(String hostname) {
-    getQueue(QUEUE).add(withUrl(PATH).method(Method.GET).header("Host", hostname));
-  }
 }

core/src/main/java/google/registry/flows/contact/ContactCheckFlow.java

@@ -56,9 +56,9 @@ public final class ContactCheckFlow implements Flow {
   @Inject ContactCheckFlow() {}
 
   @Override
-  public final EppResponse run() throws EppException {
-    extensionManager.validate();  // There are no legal extensions for this flow.
+  public EppResponse run() throws EppException {
     validateRegistrarIsLoggedIn(registrarId);
+    extensionManager.validate(); // There are no legal extensions for this flow.
     ImmutableList<String> targetIds = ((Check) resourceCommand).getTargetIds();
     verifyTargetIdCount(targetIds, maxChecks);
     ImmutableSet<String> existingIds =

core/src/main/java/google/registry/flows/contact/ContactCreateFlow.java

@@ -70,10 +70,10 @@ public final class ContactCreateFlow implements TransactionalFlow {
   @Inject ContactCreateFlow() {}
 
   @Override
-  public final EppResponse run() throws EppException {
+  public EppResponse run() throws EppException {
     extensionManager.register(MetadataExtension.class);
-    extensionManager.validate();
     validateRegistrarIsLoggedIn(registrarId);
+    extensionManager.validate();
     Create command = (Create) resourceCommand;
     DateTime now = tm().getTransactionTime();
     verifyResourceDoesNotExist(ContactResource.class, targetId, now, registrarId);

core/src/main/java/google/registry/flows/contact/ContactDeleteFlow.java

@@ -90,10 +90,10 @@ public final class ContactDeleteFlow implements TransactionalFlow {
   ContactDeleteFlow() {}
 
   @Override
-  public final EppResponse run() throws EppException {
+  public EppResponse run() throws EppException {
     extensionManager.register(MetadataExtension.class);
-    extensionManager.validate();
     validateRegistrarIsLoggedIn(registrarId);
+    extensionManager.validate();
     DateTime now = tm().getTransactionTime();
     checkLinkedDomains(targetId, now, ContactResource.class, DomainBase::getReferencedContacts);
     ContactResource existingContact = loadAndVerifyExistence(ContactResource.class, targetId, now);

core/src/main/java/google/registry/flows/contact/ContactInfoFlow.java

@@ -65,10 +65,10 @@ public final class ContactInfoFlow implements Flow {
   ContactInfoFlow() {}
 
   @Override
-  public final EppResponse run() throws EppException {
+  public EppResponse run() throws EppException {
     DateTime now = clock.nowUtc();
-    extensionManager.validate(); // There are no legal extensions for this flow.
     validateRegistrarIsLoggedIn(registrarId);
+    extensionManager.validate(); // There are no legal extensions for this flow.
     ContactResource contact = loadAndVerifyExistence(ContactResource.class, targetId, now);
     if (!isSuperuser) {
       verifyResourceOwnership(registrarId, contact);

core/src/main/java/google/registry/flows/contact/ContactTransferApproveFlow.java

@@ -74,14 +74,14 @@ public final class ContactTransferApproveFlow implements TransactionalFlow {
   @Inject ContactTransferApproveFlow() {}
 
   /**
-   * <p>The logic in this flow, which handles client approvals, very closely parallels the logic in
+   * The logic in this flow, which handles client approvals, very closely parallels the logic in
    * {@link ContactResource#cloneProjectedAtTime} which handles implicit server approvals.
    */
   @Override
-  public final EppResponse run() throws EppException {
+  public EppResponse run() throws EppException {
     extensionManager.register(MetadataExtension.class);
-    extensionManager.validate();
     validateRegistrarIsLoggedIn(registrarId);
+    extensionManager.validate();
     DateTime now = tm().getTransactionTime();
     ContactResource existingContact = loadAndVerifyExistence(ContactResource.class, targetId, now);
     verifyOptionalAuthInfo(authInfo, existingContact);

core/src/main/java/google/registry/flows/contact/ContactTransferCancelFlow.java

@@ -76,8 +76,8 @@ public final class ContactTransferCancelFlow implements TransactionalFlow {
   @Override
   public final EppResponse run() throws EppException {
     extensionManager.register(MetadataExtension.class);
-    extensionManager.validate();
     validateRegistrarIsLoggedIn(registrarId);
+    extensionManager.validate();
     DateTime now = tm().getTransactionTime();
     ContactResource existingContact = loadAndVerifyExistence(ContactResource.class, targetId, now);
     verifyOptionalAuthInfo(authInfo, existingContact);

core/src/main/java/google/registry/flows/contact/ContactTransferQueryFlow.java

@@ -63,9 +63,9 @@ public final class ContactTransferQueryFlow implements Flow {
   @Inject ContactTransferQueryFlow() {}
 
   @Override
-  public final EppResponse run() throws EppException {
-    extensionManager.validate();  // There are no legal extensions for this flow.
+  public EppResponse run() throws EppException {
     validateRegistrarIsLoggedIn(registrarId);
+    extensionManager.validate(); // There are no legal extensions for this flow.
     ContactResource contact =
         loadAndVerifyExistence(ContactResource.class, targetId, clock.nowUtc());
     verifyOptionalAuthInfo(authInfo, contact);

core/src/main/java/google/registry/flows/contact/ContactTransferRejectFlow.java

@@ -72,10 +72,10 @@ public final class ContactTransferRejectFlow implements TransactionalFlow {
   @Inject ContactTransferRejectFlow() {}
 
   @Override
-  public final EppResponse run() throws EppException {
+  public EppResponse run() throws EppException {
     extensionManager.register(MetadataExtension.class);
-    extensionManager.validate();
     validateRegistrarIsLoggedIn(registrarId);
+    extensionManager.validate();
     DateTime now = tm().getTransactionTime();
     ContactResource existingContact = loadAndVerifyExistence(ContactResource.class, targetId, now);
     verifyOptionalAuthInfo(authInfo, existingContact);

core/src/main/java/google/registry/flows/contact/ContactTransferRequestFlow.java

@@ -92,10 +92,10 @@ public final class ContactTransferRequestFlow implements TransactionalFlow {
   @Inject ContactTransferRequestFlow() {}
 
   @Override
-  public final EppResponse run() throws EppException {
+  public EppResponse run() throws EppException {
     extensionManager.register(MetadataExtension.class);
-    extensionManager.validate();
     validateRegistrarIsLoggedIn(gainingClientId);
+    extensionManager.validate();
     DateTime now = tm().getTransactionTime();
     ContactResource existingContact = loadAndVerifyExistence(ContactResource.class, targetId, now);
     verifyAuthInfoPresentForResourceTransfer(authInfo);

core/src/main/java/google/registry/flows/contact/ContactUpdateFlow.java

@@ -89,10 +89,10 @@ public final class ContactUpdateFlow implements TransactionalFlow {
   @Inject ContactUpdateFlow() {}
 
   @Override
-  public final EppResponse run() throws EppException {
+  public EppResponse run() throws EppException {
     extensionManager.register(MetadataExtension.class);
-    extensionManager.validate();
     validateRegistrarIsLoggedIn(registrarId);
+    extensionManager.validate();
     Update command = (Update) resourceCommand;
     DateTime now = tm().getTransactionTime();
     ContactResource existingContact = loadAndVerifyExistence(ContactResource.class, targetId, now);

core/src/main/java/google/registry/flows/domain/DomainCheckFlow.java

@@ -135,8 +135,8 @@ public final class DomainCheckFlow implements Flow {
     extensionManager.register(
         FeeCheckCommandExtension.class, LaunchCheckExtension.class, AllocationTokenExtension.class);
     flowCustomLogic.beforeValidation();
-    extensionManager.validate();
     validateRegistrarIsLoggedIn(registrarId);
+    extensionManager.validate();
     ImmutableList<String> domainNames = ((Check) resourceCommand).getTargetIds();
     verifyTargetIdCount(domainNames, maxChecks);
     DateTime now = clock.nowUtc();

core/src/main/java/google/registry/flows/domain/DomainClaimsCheckFlow.java

@@ -82,8 +82,8 @@ public final class DomainClaimsCheckFlow implements Flow {
   @Override
   public EppResponse run() throws EppException {
     extensionManager.register(LaunchCheckExtension.class, AllocationTokenExtension.class);
-    extensionManager.validate();
     validateRegistrarIsLoggedIn(registrarId);
+    extensionManager.validate();
     if (eppInput.getSingleExtension(AllocationTokenExtension.class).isPresent()) {
       throw new DomainClaimsCheckNotAllowedWithAllocationTokens();
     }

core/src/main/java/google/registry/flows/domain/DomainCreateFlow.java

@@ -169,6 +169,7 @@ import org.joda.time.Duration;
  * @error {@link DomainFlowUtils.FeesMismatchException}
  * @error {@link DomainFlowUtils.FeesRequiredDuringEarlyAccessProgramException}
  * @error {@link DomainFlowUtils.FeesRequiredForPremiumNameException}
+ * @error {@link DomainFlowUtils.InvalidDsRecordException}
  * @error {@link DomainFlowUtils.InvalidIdnDomainLabelException}
  * @error {@link DomainFlowUtils.InvalidPunycodeException}
  * @error {@link DomainFlowUtils.InvalidTcnIdChecksumException}
@@ -198,7 +199,7 @@ import org.joda.time.Duration;
  * @error {@link DomainPricingLogic.AllocationTokenInvalidForPremiumNameException}
  */
 @ReportingSpec(ActivityReportField.DOMAIN_CREATE)
-public class DomainCreateFlow implements TransactionalFlow {
+public final class DomainCreateFlow implements TransactionalFlow {
 
   /** Anchor tenant creates should always be for 2 years, since they get 2 years free. */
   private static final int ANCHOR_TENANT_CREATE_VALID_YEARS = 2;
@@ -219,7 +220,7 @@ public class DomainCreateFlow implements TransactionalFlow {
   @Inject DomainCreateFlow() {}
 
   @Override
-  public final EppResponse run() throws EppException {
+  public EppResponse run() throws EppException {
     extensionManager.register(
         FeeCreateCommandExtension.class,
         SecDnsCreateExtension.class,
@@ -227,9 +228,9 @@ public class DomainCreateFlow implements TransactionalFlow {
         LaunchCreateExtension.class,
         AllocationTokenExtension.class);
     flowCustomLogic.beforeValidation();
-    extensionManager.validate();
     validateRegistrarIsLoggedIn(registrarId);
     verifyRegistrarIsActive(registrarId);
+    extensionManager.validate();
     DateTime now = tm().getTransactionTime();
     DomainCommand.Create command = cloneAndLinkReferences((Create) resourceCommand, now);
     Period period = command.getPeriod();

core/src/main/java/google/registry/flows/domain/DomainDeleteFlow.java

@@ -138,12 +138,12 @@ public final class DomainDeleteFlow implements TransactionalFlow {
   @Inject DomainDeleteFlow() {}
 
   @Override
-  public final EppResponse run() throws EppException {
+  public EppResponse run() throws EppException {
     extensionManager.register(
         MetadataExtension.class, SecDnsCreateExtension.class, DomainDeleteSuperuserExtension.class);
     flowCustomLogic.beforeValidation();
-    extensionManager.validate();
     validateRegistrarIsLoggedIn(registrarId);
+    extensionManager.validate();
     DateTime now = tm().getTransactionTime();
     // Loads the target resource if it exists
     DomainBase existingDomain = loadAndVerifyExistence(DomainBase.class, targetId, now);

core/src/main/java/google/registry/flows/domain/DomainFlowUtils.java

@@ -129,6 +129,7 @@ import google.registry.model.tld.label.ReservedList;
 import google.registry.model.tmch.ClaimsListDao;
 import google.registry.persistence.VKey;
 import google.registry.tldconfig.idn.IdnLabelValidator;
+import google.registry.tools.DigestType;
 import google.registry.util.Idn;
 import java.math.BigDecimal;
 import java.util.Collection;
@@ -144,6 +145,7 @@ import org.joda.money.CurrencyUnit;
 import org.joda.money.Money;
 import org.joda.time.DateTime;
 import org.joda.time.Duration;
+import org.xbill.DNS.DNSSEC.Algorithm;
 
 /** Static utility functions for domain flows. */
 public class DomainFlowUtils {
@@ -293,13 +295,46 @@ public class DomainFlowUtils {
 
   /** Check that the DS data that will be set on a domain is valid. */
   static void validateDsData(Set<DelegationSignerData> dsData) throws EppException {
-    if (dsData != null && dsData.size() > MAX_DS_RECORDS_PER_DOMAIN) {
-      throw new TooManyDsRecordsException(
-          String.format(
-              "A maximum of %s DS records are allowed per domain.", MAX_DS_RECORDS_PER_DOMAIN));
+    if (dsData != null) {
+      if (dsData.size() > MAX_DS_RECORDS_PER_DOMAIN) {
+        throw new TooManyDsRecordsException(
+            String.format(
+                "A maximum of %s DS records are allowed per domain.", MAX_DS_RECORDS_PER_DOMAIN));
+      }
+      // TODO(sarahbot@): Add signature length verification
+      ImmutableList<DelegationSignerData> invalidAlgorithms =
+          dsData.stream()
+              .filter(ds -> !validateAlgorithm(ds.getAlgorithm()))
+              .collect(toImmutableList());
+      if (!invalidAlgorithms.isEmpty()) {
+        throw new InvalidDsRecordException(
+            String.format(
+                "Domain contains DS record(s) with an invalid algorithm wire value: %s",
+                invalidAlgorithms));
+      }
+      ImmutableList<DelegationSignerData> invalidDigestTypes =
+          dsData.stream()
+              .filter(ds -> !DigestType.fromWireValue(ds.getDigestType()).isPresent())
+              .collect(toImmutableList());
+      if (!invalidDigestTypes.isEmpty()) {
+        throw new InvalidDsRecordException(
+            String.format(
+                "Domain contains DS record(s) with an invalid digest type: %s",
+                invalidDigestTypes));
+      }
     }
   }
 
+  public static boolean validateAlgorithm(int alg) {
+    if (alg > 255 || alg < 0) {
+      return false;
+    }
+    // Algorithms that are reserved or unassigned will just return a string representation of their
+    // integer wire value.
+    String algorithm = Algorithm.string(alg);
+    return !algorithm.equals(Integer.toString(alg));
+  }
+
   /** We only allow specifying years in a period. */
   static Period verifyUnitIsYears(Period period) throws EppException {
     if (!checkNotNull(period).getUnit().equals(Period.Unit.YEARS)) {
@@ -1217,6 +1252,13 @@ public class DomainFlowUtils {
     }
   }
 
+  /** Domain has an invalid DS record. */
+  static class InvalidDsRecordException extends ParameterValuePolicyErrorException {
+    public InvalidDsRecordException(String message) {
+      super(message);
+    }
+  }
+
   /** Domain name is under tld which doesn't exist. */
   static class TldDoesNotExistException extends ParameterValueRangeErrorException {
     public TldDoesNotExistException(String tld) {

core/src/main/java/google/registry/flows/domain/DomainInfoFlow.java

@@ -91,11 +91,11 @@ public final class DomainInfoFlow implements Flow {
   DomainInfoFlow() {}
 
   @Override
-  public final EppResponse run() throws EppException {
+  public EppResponse run() throws EppException {
     extensionManager.register(FeeInfoCommandExtensionV06.class);
     flowCustomLogic.beforeValidation();
-    extensionManager.validate();
     validateRegistrarIsLoggedIn(registrarId);
+    extensionManager.validate();
     DateTime now = clock.nowUtc();
     DomainBase domain = verifyExistence(
         DomainBase.class, targetId, loadByForeignKey(DomainBase.class, targetId, now));

core/src/main/java/google/registry/flows/domain/DomainRenewFlow.java

@@ -135,12 +135,12 @@ public final class DomainRenewFlow implements TransactionalFlow {
   @Inject DomainRenewFlow() {}
 
   @Override
-  public final EppResponse run() throws EppException {
+  public EppResponse run() throws EppException {
     extensionManager.register(FeeRenewCommandExtension.class, MetadataExtension.class);
     flowCustomLogic.beforeValidation();
-    extensionManager.validate();
     validateRegistrarIsLoggedIn(registrarId);
     verifyRegistrarIsActive(registrarId);
+    extensionManager.validate();
     DateTime now = tm().getTransactionTime();
     Renew command = (Renew) resourceCommand;
     // Loads the target resource if it exists

core/src/main/java/google/registry/flows/domain/DomainRestoreRequestFlow.java

@@ -128,14 +128,14 @@ public final class DomainRestoreRequestFlow implements TransactionalFlow {
   @Inject DomainRestoreRequestFlow() {}
 
   @Override
-  public final EppResponse run() throws EppException {
+  public EppResponse run() throws EppException {
     extensionManager.register(
         FeeUpdateCommandExtension.class,
         MetadataExtension.class,
         RgpUpdateExtension.class);
-    extensionManager.validate();
     validateRegistrarIsLoggedIn(registrarId);
     verifyRegistrarIsActive(registrarId);
+    extensionManager.validate();
     Update command = (Update) resourceCommand;
     DateTime now = tm().getTransactionTime();
     DomainBase existingDomain = loadAndVerifyExistence(DomainBase.class, targetId, now);

core/src/main/java/google/registry/flows/domain/DomainTransferApproveFlow.java

@@ -99,14 +99,14 @@ public final class DomainTransferApproveFlow implements TransactionalFlow {
   @Inject DomainTransferApproveFlow() {}
 
   /**
-   * <p>The logic in this flow, which handles client approvals, very closely parallels the logic in
+   * The logic in this flow, which handles client approvals, very closely parallels the logic in
    * {@link DomainBase#cloneProjectedAtTime} which handles implicit server approvals.
    */
   @Override
-  public final EppResponse run() throws EppException {
+  public EppResponse run() throws EppException {
     extensionManager.register(MetadataExtension.class);
-    extensionManager.validate();
     validateRegistrarIsLoggedIn(registrarId);
+    extensionManager.validate();
     DateTime now = tm().getTransactionTime();
     DomainBase existingDomain = loadAndVerifyExistence(DomainBase.class, targetId, now);
     verifyOptionalAuthInfo(authInfo, existingDomain);

core/src/main/java/google/registry/flows/domain/DomainTransferCancelFlow.java

@@ -86,10 +86,10 @@ public final class DomainTransferCancelFlow implements TransactionalFlow {
   @Inject DomainTransferCancelFlow() {}
 
   @Override
-  public final EppResponse run() throws EppException {
+  public EppResponse run() throws EppException {
     extensionManager.register(MetadataExtension.class);
-    extensionManager.validate();
     validateRegistrarIsLoggedIn(registrarId);
+    extensionManager.validate();
     DateTime now = tm().getTransactionTime();
     DomainBase existingDomain = loadAndVerifyExistence(DomainBase.class, targetId, now);
     verifyOptionalAuthInfo(authInfo, existingDomain);

core/src/main/java/google/registry/flows/domain/DomainTransferQueryFlow.java

@@ -67,9 +67,9 @@ public final class DomainTransferQueryFlow implements Flow {
   @Inject DomainTransferQueryFlow() {}
 
   @Override
-  public final EppResponse run() throws EppException {
-    extensionManager.validate();  // There are no legal extensions for this flow.
+  public EppResponse run() throws EppException {
     validateRegistrarIsLoggedIn(registrarId);
+    extensionManager.validate(); // There are no legal extensions for this flow.
     DateTime now = clock.nowUtc();
     DomainBase domain = loadAndVerifyExistence(DomainBase.class, targetId, now);
     verifyOptionalAuthInfo(authInfo, domain);

core/src/main/java/google/registry/flows/domain/DomainTransferRejectFlow.java

@@ -88,10 +88,10 @@ public final class DomainTransferRejectFlow implements TransactionalFlow {
   @Inject DomainTransferRejectFlow() {}
 
   @Override
-  public final EppResponse run() throws EppException {
+  public EppResponse run() throws EppException {
     extensionManager.register(MetadataExtension.class);
-    extensionManager.validate();
     validateRegistrarIsLoggedIn(registrarId);
+    extensionManager.validate();
     DateTime now = tm().getTransactionTime();
     DomainBase existingDomain = loadAndVerifyExistence(DomainBase.class, targetId, now);
     Registry registry = Registry.get(existingDomain.getTld());

core/src/main/java/google/registry/flows/domain/DomainTransferRequestFlow.java

@@ -139,14 +139,14 @@ public final class DomainTransferRequestFlow implements TransactionalFlow {
   @Inject DomainTransferRequestFlow() {}
 
   @Override
-  public final EppResponse run() throws EppException {
+  public EppResponse run() throws EppException {
     extensionManager.register(
         DomainTransferRequestSuperuserExtension.class,
         FeeTransferCommandExtension.class,
         MetadataExtension.class);
-    extensionManager.validate();
     validateRegistrarIsLoggedIn(gainingClientId);
     verifyRegistrarIsActive(gainingClientId);
+    extensionManager.validate();
     DateTime now = tm().getTransactionTime();
     DomainBase existingDomain = loadAndVerifyExistence(DomainBase.class, targetId, now);
     Optional<DomainTransferRequestSuperuserExtension> superuserExtension =

core/src/main/java/google/registry/flows/domain/DomainUpdateFlow.java

@@ -43,6 +43,7 @@ import static google.registry.flows.domain.DomainFlowUtils.verifyNotInPendingDel
 import static google.registry.model.reporting.HistoryEntry.Type.DOMAIN_UPDATE;
 import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
 
+import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.ImmutableSortedSet;
 import com.google.common.collect.Ordering;
@@ -77,9 +78,11 @@ import google.registry.model.domain.secdns.SecDnsUpdateExtension;
 import google.registry.model.domain.superuser.DomainUpdateSuperuserExtension;
 import google.registry.model.eppcommon.AuthInfo;
 import google.registry.model.eppcommon.StatusValue;
+import google.registry.model.eppcommon.Trid;
 import google.registry.model.eppinput.EppInput;
 import google.registry.model.eppinput.ResourceCommand;
 import google.registry.model.eppoutput.EppResponse;
+import google.registry.model.poll.PendingActionNotificationResponse.DomainPendingActionNotificationResponse;
 import google.registry.model.poll.PollMessage;
 import google.registry.model.reporting.IcannReportingTypes.ActivityReportField;
 import google.registry.model.tld.Registry;
@@ -111,6 +114,7 @@ import org.joda.time.DateTime;
  * @error {@link DomainFlowUtils.EmptySecDnsUpdateException}
  * @error {@link DomainFlowUtils.FeesMismatchException}
  * @error {@link DomainFlowUtils.FeesRequiredForNonFreeOperationException}
+ * @error {@link DomainFlowUtils.InvalidDsRecordException}
  * @error {@link DomainFlowUtils.LinkedResourcesDoNotExistException}
  * @error {@link DomainFlowUtils.LinkedResourceInPendingDeleteProhibitsOperationException}
  * @error {@link DomainFlowUtils.MaxSigLifeChangeNotSupportedException}
@@ -149,6 +153,7 @@ public final class DomainUpdateFlow implements TransactionalFlow {
   @Inject @RegistrarId String registrarId;
   @Inject @TargetId String targetId;
   @Inject @Superuser boolean isSuperuser;
+  @Inject Trid trid;
   @Inject DomainHistory.Builder historyBuilder;
   @Inject DnsQueue dnsQueue;
   @Inject EppResponse.Builder responseBuilder;
@@ -164,8 +169,8 @@ public final class DomainUpdateFlow implements TransactionalFlow {
         SecDnsUpdateExtension.class,
         DomainUpdateSuperuserExtension.class);
     flowCustomLogic.beforeValidation();
-    extensionManager.validate();
     validateRegistrarIsLoggedIn(registrarId);
+    extensionManager.validate();
     DateTime now = tm().getTransactionTime();
     Update command = cloneAndLinkReferences((Update) resourceCommand, now);
     DomainBase existingDomain = loadAndVerifyExistence(DomainBase.class, targetId, now);
@@ -360,6 +365,10 @@ public final class DomainUpdateFlow implements TransactionalFlow {
             .setEventTime(now)
             .setRegistrarId(existingDomain.getCurrentSponsorRegistrarId())
             .setMsg(msg)
+            .setResponseData(
+                ImmutableList.of(
+                    DomainPendingActionNotificationResponse.create(
+                        existingDomain.getDomainName(), true, trid, now)))
             .build());
   }
 }

core/src/main/java/google/registry/flows/host/HostCheckFlow.java

@@ -56,9 +56,9 @@ public final class HostCheckFlow implements Flow {
   @Inject HostCheckFlow() {}
 
   @Override
-  public final EppResponse run() throws EppException {
-    extensionManager.validate();  // There are no legal extensions for this flow.
+  public EppResponse run() throws EppException {
     validateRegistrarIsLoggedIn(registrarId);
+    extensionManager.validate(); // There are no legal extensions for this flow.
     ImmutableList<String> hostnames = ((Check) resourceCommand).getTargetIds();
     verifyTargetIdCount(hostnames, maxChecks);
     ImmutableSet<String> existingIds =

core/src/main/java/google/registry/flows/host/HostCreateFlow.java

@@ -100,10 +100,10 @@ public final class HostCreateFlow implements TransactionalFlow {
   HostCreateFlow() {}
 
   @Override
-  public final EppResponse run() throws EppException {
+  public EppResponse run() throws EppException {
     extensionManager.register(MetadataExtension.class);
-    extensionManager.validate();
     validateRegistrarIsLoggedIn(registrarId);
+    extensionManager.validate();
     Create command = (Create) resourceCommand;
     DateTime now = tm().getTransactionTime();
     verifyResourceDoesNotExist(HostResource.class, targetId, now, registrarId);

core/src/main/java/google/registry/flows/host/HostDeleteFlow.java

@@ -92,10 +92,10 @@ public final class HostDeleteFlow implements TransactionalFlow {
   HostDeleteFlow() {}
 
   @Override
-  public final EppResponse run() throws EppException {
+  public EppResponse run() throws EppException {
     extensionManager.register(MetadataExtension.class);
-    extensionManager.validate();
     validateRegistrarIsLoggedIn(registrarId);
+    extensionManager.validate();
     DateTime now = tm().getTransactionTime();
     validateHostName(targetId);
     checkLinkedDomains(targetId, now, HostResource.class, DomainBase::getNameservers);

core/src/main/java/google/registry/flows/host/HostInfoFlow.java

@@ -62,8 +62,8 @@ public final class HostInfoFlow implements Flow {
 
   @Override
   public EppResponse run() throws EppException {
-    extensionManager.validate();  // There are no legal extensions for this flow.
     validateRegistrarIsLoggedIn(registrarId);
+    extensionManager.validate(); // There are no legal extensions for this flow.
     validateHostName(targetId);
     DateTime now = clock.nowUtc();
     HostResource host = loadAndVerifyExistence(HostResource.class, targetId, now);

core/src/main/java/google/registry/flows/host/HostUpdateFlow.java

@@ -125,10 +125,10 @@ public final class HostUpdateFlow implements TransactionalFlow {
   @Inject HostUpdateFlow() {}
 
   @Override
-  public final EppResponse run() throws EppException {
+  public EppResponse run() throws EppException {
     extensionManager.register(MetadataExtension.class);
-    extensionManager.validate();
     validateRegistrarIsLoggedIn(registrarId);
+    extensionManager.validate();
     Update command = (Update) resourceCommand;
     Change change = command.getInnerChange();
     String suppliedNewHostName = change.getFullyQualifiedHostName();

core/src/main/java/google/registry/flows/poll/PollAckFlow.java

@@ -45,17 +45,17 @@ import org.joda.time.DateTime;
 /**
  * An EPP flow for acknowledging {@link PollMessage}s.
  *
- * <p>Registrars refer to poll messages using an externally visible id generated by
- * {@link PollMessageExternalKeyConverter}. One-time poll messages are deleted from Datastore once
- * they are ACKed, whereas autorenew poll messages are simply marked as read, and won't be delivered
- * again until the next year of their recurrence.
+ * <p>Registrars refer to poll messages using an externally visible id generated by {@link
+ * PollMessageExternalKeyConverter}. One-time poll messages are deleted from Datastore once they are
+ * ACKed, whereas autorenew poll messages are simply marked as read, and won't be delivered again
+ * until the next year of their recurrence.
  *
  * @error {@link PollAckFlow.InvalidMessageIdException}
  * @error {@link PollAckFlow.MessageDoesNotExistException}
  * @error {@link PollAckFlow.MissingMessageIdException}
  * @error {@link PollAckFlow.NotAuthorizedToAckMessageException}
  */
-public class PollAckFlow implements TransactionalFlow {
+public final class PollAckFlow implements TransactionalFlow {
 
   @Inject ExtensionManager extensionManager;
   @Inject @RegistrarId String registrarId;
@@ -65,9 +65,9 @@ public class PollAckFlow implements TransactionalFlow {
   @Inject PollAckFlow() {}
 
   @Override
-  public final EppResponse run() throws EppException {
-    extensionManager.validate();  // There are no legal extensions for this flow.
+  public EppResponse run() throws EppException {
     validateRegistrarIsLoggedIn(registrarId);
+    extensionManager.validate(); // There are no legal extensions for this flow.
     if (messageId.isEmpty()) {
       throw new MissingMessageIdException();
     }

core/src/main/java/google/registry/flows/poll/PollRequestFlow.java

@@ -47,7 +47,7 @@ import org.joda.time.DateTime;
  *
  * @error {@link PollRequestFlow.UnexpectedMessageIdException}
  */
-public class PollRequestFlow implements Flow {
+public final class PollRequestFlow implements Flow {
 
   @Inject ExtensionManager extensionManager;
   @Inject @RegistrarId String registrarId;
@@ -57,9 +57,9 @@ public class PollRequestFlow implements Flow {
   @Inject PollRequestFlow() {}
 
   @Override
-  public final EppResponse run() throws EppException {
-    extensionManager.validate();  // There are no legal extensions for this flow.
+  public EppResponse run() throws EppException {
     validateRegistrarIsLoggedIn(registrarId);
+    extensionManager.validate(); // There are no legal extensions for this flow.
     if (!messageId.isEmpty()) {
       throw new UnexpectedMessageIdException();
     }

core/src/main/java/google/registry/flows/session/HelloFlow.java

@@ -27,7 +27,7 @@ import javax.inject.Inject;
  *
  * @error {@link google.registry.flows.FlowUtils.GenericXmlSyntaxErrorException}
  */
-public class HelloFlow implements Flow {
+public final class HelloFlow implements Flow {
 
   @Inject ExtensionManager extensionManager;
   @Inject Clock clock;

core/src/main/java/google/registry/flows/session/LoginFlow.java

@@ -90,7 +90,7 @@ public class LoginFlow implements Flow {
   }
 
   /** Run the flow without bothering to log errors. The {@link #run} method will do that for us. */
-  private final EppResponse runWithoutLogging() throws EppException {
+  private EppResponse runWithoutLogging() throws EppException {
     extensionManager.validate();  // There are no legal extensions for this flow.
     Login login = (Login) eppInput.getCommandWrapper().getCommand();
     if (!registrarId.isEmpty()) {

core/src/main/java/google/registry/flows/session/LogoutFlow.java

@@ -30,7 +30,7 @@ import javax.inject.Inject;
  *
  * @error {@link google.registry.flows.FlowUtils.NotLoggedInException}
  */
-public class LogoutFlow implements Flow {
+public final class LogoutFlow implements Flow {
 
   @Inject ExtensionManager extensionManager;
   @Inject @RegistrarId String registrarId;
@@ -39,9 +39,9 @@ public class LogoutFlow implements Flow {
   @Inject LogoutFlow() {}
 
   @Override
-  public final EppResponse run() throws EppException {
-    extensionManager.validate();  // There are no legal extensions for this flow.
+  public EppResponse run() throws EppException {
     validateRegistrarIsLoggedIn(registrarId);
+    extensionManager.validate(); // There are no legal extensions for this flow.
     sessionMetadata.invalidate();
     return responseBuilder.setResultFromCode(SUCCESS_AND_CLOSE).build();
   }

core/src/main/java/google/registry/loadtest/LoadTestAction.java

@@ -14,8 +14,6 @@
 
 package google.registry.loadtest;
 
-import static com.google.appengine.api.taskqueue.QueueConstants.maxTasksPerAdd;
-import static com.google.appengine.api.taskqueue.QueueFactory.getQueue;
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.collect.ImmutableList.toImmutableList;
 import static com.google.common.collect.Lists.partition;
@@ -24,16 +22,20 @@ import static google.registry.util.ResourceUtils.readResourceUtf8;
 import static java.util.Arrays.asList;
 import static org.joda.time.DateTimeZone.UTC;
 
-import com.google.appengine.api.taskqueue.TaskOptions;
+import com.google.cloud.tasks.v2.Task;
 import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMultimap;
 import com.google.common.collect.Iterators;
 import com.google.common.flogger.FluentLogger;
+import com.google.protobuf.Timestamp;
 import google.registry.config.RegistryEnvironment;
 import google.registry.request.Action;
+import google.registry.request.Action.Service;
 import google.registry.request.Parameter;
 import google.registry.request.auth.Auth;
 import google.registry.security.XsrfTokenManager;
-import google.registry.util.TaskQueueUtils;
+import google.registry.util.CloudTasksUtils;
+import java.time.Instant;
 import java.util.Arrays;
 import java.util.Iterator;
 import java.util.List;
@@ -62,6 +64,7 @@ public class LoadTestAction implements Runnable {
   private static final FluentLogger logger = FluentLogger.forEnclosingClass();
 
   private static final int NUM_QUEUES = 10;
+  private static final int MAX_TASKS_PER_LOAD = 100;
   private static final int ARBITRARY_VALID_HOST_LENGTH = 40;
   private static final int MAX_CONTACT_LENGTH = 13;
   private static final int MAX_DOMAIN_LABEL_LENGTH = 63;
@@ -146,7 +149,7 @@ public class LoadTestAction implements Runnable {
   @Parameter("hostInfos")
   int hostInfosPerSecond;
 
-  @Inject TaskQueueUtils taskQueueUtils;
+  @Inject CloudTasksUtils cloudTasksUtils;
 
   private final String xmlContactCreateTmpl;
   private final String xmlContactCreateFail;
@@ -208,7 +211,7 @@ public class LoadTestAction implements Runnable {
     ImmutableList<String> contactNames = contactNamesBuilder.build();
     ImmutableList<String> hostPrefixes = hostPrefixesBuilder.build();
 
-    ImmutableList.Builder<TaskOptions> tasks = new ImmutableList.Builder<>();
+    ImmutableList.Builder<Task> tasks = new ImmutableList.Builder<>();
     for (int offsetSeconds = 0; offsetSeconds < runSeconds; offsetSeconds++) {
       DateTime startSecond = initialStartSecond.plusSeconds(offsetSeconds);
       // The first "failed" creates might actually succeed if the object doesn't already exist, but
@@ -254,7 +257,7 @@ public class LoadTestAction implements Runnable {
                   .collect(toImmutableList()),
               startSecond));
     }
-    ImmutableList<TaskOptions> taskOptions = tasks.build();
+    ImmutableList<Task> taskOptions = tasks.build();
     enqueue(taskOptions);
     logger.atInfo().log("Added %d total load test tasks.", taskOptions.size());
   }
@@ -322,28 +325,50 @@ public class LoadTestAction implements Runnable {
     return name.toString();
   }
 
-  private List<TaskOptions> createTasks(List<String> xmls, DateTime start) {
-    ImmutableList.Builder<TaskOptions> tasks = new ImmutableList.Builder<>();
+  private List<Task> createTasks(List<String> xmls, DateTime start) {
+    ImmutableList.Builder<Task> tasks = new ImmutableList.Builder<>();
     for (int i = 0; i < xmls.size(); i++) {
       // Space tasks evenly within across a second.
-      int offsetMillis = (int) (1000.0 / xmls.size() * i);
+      Instant scheduleTime =
+          Instant.ofEpochMilli(start.plusMillis((int) (1000.0 / xmls.size() * i)).getMillis());
       tasks.add(
-          TaskOptions.Builder.withUrl("/_dr/epptool")
-              .etaMillis(start.getMillis() + offsetMillis)
-              .header(X_CSRF_TOKEN, xsrfToken)
-              .param("clientId", registrarId)
-              .param("superuser", Boolean.FALSE.toString())
-              .param("dryRun", Boolean.FALSE.toString())
-              .param("xml", xmls.get(i)));
+          Task.newBuilder()
+              .setAppEngineHttpRequest(
+                  CloudTasksUtils.createPostTask(
+                          "/_dr/epptool",
+                          Service.TOOLS.toString(),
+                          ImmutableMultimap.of(
+                              "clientId",
+                              registrarId,
+                              "superuser",
+                              Boolean.FALSE.toString(),
+                              "dryRun",
+                              Boolean.FALSE.toString(),
+                              "xml",
+                              xmls.get(i)))
+                      .toBuilder()
+                      .getAppEngineHttpRequest()
+                      .toBuilder()
+                      // instead of adding the X_CSRF_TOKEN to params, this remains as part of
+                      // headers because of the existing setup for authentication in {@link
+                      // google.registry.request.auth.LegacyAuthenticationMechanism}
+                      .putHeaders(X_CSRF_TOKEN, xsrfToken)
+                      .build())
+              .setScheduleTime(
+                  Timestamp.newBuilder()
+                      .setSeconds(scheduleTime.getEpochSecond())
+                      .setNanos(scheduleTime.getNano())
+                      .build())
+              .build());
     }
     return tasks.build();
   }
 
-  private void enqueue(List<TaskOptions> tasks) {
-    List<List<TaskOptions>> chunks = partition(tasks, maxTasksPerAdd());
+  private void enqueue(List<Task> tasks) {
+    List<List<Task>> chunks = partition(tasks, MAX_TASKS_PER_LOAD);
     // Farm out tasks to multiple queues to work around queue qps quotas.
     for (int i = 0; i < chunks.size(); i++) {
-      taskQueueUtils.enqueue(getQueue("load" + (i % NUM_QUEUES)), chunks.get(i));
+      cloudTasksUtils.enqueue("load" + (i % NUM_QUEUES), chunks.get(i));
     }
   }
 }

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

@@ -15,6 +15,7 @@
 package google.registry.model;
 
 import com.google.common.collect.ImmutableSet;
+import google.registry.model.annotations.DeleteAfterMigration;
 import google.registry.model.billing.BillingEvent;
 import google.registry.model.common.Cursor;
 import google.registry.model.common.EntityGroupRoot;
@@ -45,6 +46,7 @@ import google.registry.model.server.ServerSecret;
 import google.registry.model.tld.Registry;
 
 /** Sets of classes of the Objectify-registered entities in use throughout the model. */
+@DeleteAfterMigration
 public final class EntityClasses {
 
   /** Set of entity classes. */

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

@@ -21,6 +21,7 @@ import static com.google.common.collect.Sets.union;
 import static google.registry.config.RegistryConfig.getEppResourceCachingDuration;
 import static google.registry.config.RegistryConfig.getEppResourceMaxCachedEntries;
 import static google.registry.persistence.transaction.TransactionManagerFactory.ofyTm;
+import static google.registry.persistence.transaction.TransactionManagerFactory.replicaTm;
 import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
 import static google.registry.util.CollectionUtils.nullToEmpty;
 import static google.registry.util.CollectionUtils.nullToEmptyImmutableCopy;
@@ -380,13 +381,13 @@ public abstract class EppResource extends BackupGroupRoot implements Buildable {
 
         @Override
         public EppResource load(VKey<? extends EppResource> key) {
-          return tm().doTransactionless(() -> tm().loadByKey(key));
+          return replicaTm().doTransactionless(() -> replicaTm().loadByKey(key));
         }
 
         @Override
         public Map<VKey<? extends EppResource>, EppResource> loadAll(
             Iterable<? extends VKey<? extends EppResource>> keys) {
-          return tm().doTransactionless(() -> tm().loadByKeys(keys));
+          return replicaTm().doTransactionless(() -> replicaTm().loadByKeys(keys));
         }
       };
 

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

@@ -20,6 +20,7 @@ import com.google.appengine.api.datastore.DatastoreServiceFactory;
 import com.google.common.annotations.VisibleForTesting;
 import google.registry.beam.common.RegistryPipelineWorkerInitializer;
 import google.registry.config.RegistryEnvironment;
+import google.registry.model.annotations.DeleteAfterMigration;
 import java.util.concurrent.atomic.AtomicLong;
 
 /**
@@ -28,6 +29,7 @@ import java.util.concurrent.atomic.AtomicLong;
  * <p>In non-test, non-beam environments the Id is generated by Datastore, otherwise it's from an
  * atomic long number that's incremented every time this method is called.
  */
+@DeleteAfterMigration
 public final class IdService {
 
   /**

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

@@ -19,12 +19,14 @@ import static com.google.common.base.Predicates.subtypeOf;
 import static java.util.stream.Collectors.joining;
 
 import com.google.common.collect.Ordering;
+import google.registry.model.annotations.DeleteAfterMigration;
 import java.util.ArrayDeque;
 import java.util.Queue;
 import java.util.SortedSet;
 import java.util.TreeSet;
 
 /** Utility methods for getting the version of the model schema from the model code. */
+@DeleteAfterMigration
 public final class SchemaVersion {
 
   /**

core/src/main/java/google/registry/model/common/CrossTldSingleton.java

@@ -20,11 +20,13 @@ import com.googlecode.objectify.Key;
 import com.googlecode.objectify.annotation.Id;
 import com.googlecode.objectify.annotation.Parent;
 import google.registry.model.ImmutableObject;
+import google.registry.model.annotations.DeleteAfterMigration;
 import google.registry.model.annotations.InCrossTld;
 import javax.persistence.MappedSuperclass;
 import javax.persistence.Transient;
 
 /** A singleton entity in Datastore. */
+@DeleteAfterMigration
 @MappedSuperclass
 @InCrossTld
 public abstract class CrossTldSingleton extends ImmutableObject {

core/src/main/java/google/registry/model/common/DatabaseMigrationStateSchedule.java

@@ -26,6 +26,7 @@ import com.google.common.collect.ImmutableMultimap;
 import com.google.common.collect.ImmutableSortedMap;
 import com.google.common.flogger.FluentLogger;
 import google.registry.config.RegistryEnvironment;
+import google.registry.model.annotations.DeleteAfterMigration;
 import google.registry.model.common.TimedTransitionProperty.TimedTransition;
 import google.registry.model.replay.SqlOnlyEntity;
 import java.time.Duration;
@@ -39,6 +40,7 @@ import org.joda.time.DateTime;
  * <p>The entity is stored in SQL throughout the entire migration so as to have a single point of
  * access.
  */
+@DeleteAfterMigration
 @Entity
 public class DatabaseMigrationStateSchedule extends CrossTldSingleton implements SqlOnlyEntity {
 

core/src/main/java/google/registry/model/common/EntityGroupRoot.java

@@ -19,6 +19,7 @@ import com.googlecode.objectify.Key;
 import com.googlecode.objectify.annotation.Entity;
 import com.googlecode.objectify.annotation.Id;
 import google.registry.model.BackupGroupRoot;
+import google.registry.model.annotations.DeleteAfterMigration;
 import google.registry.model.replay.DatastoreOnlyEntity;
 import javax.annotation.Nullable;
 
@@ -37,6 +38,7 @@ import javax.annotation.Nullable;
  * entity group for the single namespace where global data applicable for all TLDs lived.
  */
 @Entity
+@DeleteAfterMigration
 public class EntityGroupRoot extends BackupGroupRoot implements DatastoreOnlyEntity {
 
   @SuppressWarnings("unused")

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

@@ -34,6 +34,9 @@ import javax.persistence.AccessType;
 @ReportedOn
 @Entity
 @javax.persistence.Entity(name = "Host")
+@javax.persistence.Table(
+    name = "Host",
+    indexes = {@javax.persistence.Index(columnList = "hostName")})
 @ExternalMessagingName("host")
 @WithStringVKey
 @Access(AccessType.FIELD) // otherwise it'll use the default if the repoId (property)

core/src/main/java/google/registry/model/index/EppResourceIndex.java

@@ -24,12 +24,15 @@ import com.googlecode.objectify.annotation.Index;
 import com.googlecode.objectify.annotation.Parent;
 import google.registry.model.BackupGroupRoot;
 import google.registry.model.EppResource;
+import google.registry.model.annotations.DeleteAfterMigration;
 import google.registry.model.annotations.ReportedOn;
 import google.registry.model.replay.DatastoreOnlyEntity;
+import google.registry.persistence.VKey;
 
 /** An index that allows for quick enumeration of all EppResource entities (e.g. via map reduce). */
 @ReportedOn
 @Entity
+@DeleteAfterMigration
 public class EppResourceIndex extends BackupGroupRoot implements DatastoreOnlyEntity {
 
   @Id String id;
@@ -64,8 +67,9 @@ public class EppResourceIndex extends BackupGroupRoot implements DatastoreOnlyEn
     EppResourceIndex instance = instantiate(EppResourceIndex.class);
     instance.reference = resourceKey;
     instance.kind = resourceKey.getKind();
-    // TODO(b/207368050): figure out if this value has ever been used other than test cases
-    instance.id = resourceKey.getString(); // creates a web-safe key string
+    // creates a web-safe key string, this value is never used
+    // TODO(b/211785379): remove unused id
+    instance.id = VKey.from(resourceKey).stringify();
     instance.bucket = bucket;
     return instance;
   }

core/src/main/java/google/registry/model/index/EppResourceIndexBucket.java

@@ -23,12 +23,14 @@ import com.googlecode.objectify.annotation.Entity;
 import com.googlecode.objectify.annotation.Id;
 import google.registry.model.EppResource;
 import google.registry.model.ImmutableObject;
+import google.registry.model.annotations.DeleteAfterMigration;
 import google.registry.model.annotations.VirtualEntity;
 import google.registry.model.replay.DatastoreOnlyEntity;
 
 /** A virtual entity to represent buckets to which EppResourceIndex objects are randomly added. */
 @Entity
 @VirtualEntity
+@DeleteAfterMigration
 public class EppResourceIndexBucket extends ImmutableObject implements DatastoreOnlyEntity {
 
   @SuppressWarnings("unused")

core/src/main/java/google/registry/model/index/ForeignKeyIndex.java

@@ -21,6 +21,7 @@ import static google.registry.config.RegistryConfig.getEppResourceCachingDuratio
 import static google.registry.config.RegistryConfig.getEppResourceMaxCachedEntries;
 import static google.registry.model.ofy.ObjectifyService.auditedOfy;
 import static google.registry.persistence.transaction.TransactionManagerFactory.jpaTm;
+import static google.registry.persistence.transaction.TransactionManagerFactory.replicaJpaTm;
 import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
 import static google.registry.util.CollectionUtils.entriesToImmutableMap;
 import static google.registry.util.TypeUtils.instantiate;
@@ -43,6 +44,7 @@ import com.googlecode.objectify.annotation.Index;
 import google.registry.config.RegistryConfig;
 import google.registry.model.BackupGroupRoot;
 import google.registry.model.EppResource;
+import google.registry.model.annotations.DeleteAfterMigration;
 import google.registry.model.annotations.ReportedOn;
 import google.registry.model.contact.ContactResource;
 import google.registry.model.domain.DomainBase;
@@ -50,6 +52,7 @@ import google.registry.model.host.HostResource;
 import google.registry.model.replay.DatastoreOnlyEntity;
 import google.registry.persistence.VKey;
 import google.registry.persistence.transaction.CriteriaQueryBuilder;
+import google.registry.persistence.transaction.JpaTransactionManager;
 import google.registry.util.NonFinalForTesting;
 import java.util.Collection;
 import java.util.Comparator;
@@ -65,6 +68,7 @@ import org.joda.time.Duration;
  * the foreign key string. The instance is never deleted, but it is updated if a newer entity
  * becomes the active entity.
  */
+@DeleteAfterMigration
 public abstract class ForeignKeyIndex<E extends EppResource> extends BackupGroupRoot {
 
   /** The {@link ForeignKeyIndex} type for {@link ContactResource} entities. */
@@ -196,7 +200,7 @@ public abstract class ForeignKeyIndex<E extends EppResource> extends BackupGroup
    */
   public static <E extends EppResource> ImmutableMap<String, ForeignKeyIndex<E>> load(
       Class<E> clazz, Collection<String> foreignKeys, final DateTime now) {
-    return loadIndexesFromStore(clazz, foreignKeys, true).entrySet().stream()
+    return loadIndexesFromStore(clazz, foreignKeys, true, false).entrySet().stream()
         .filter(e -> now.isBefore(e.getValue().getDeletionTime()))
         .collect(entriesToImmutableMap());
   }
@@ -215,7 +219,10 @@ public abstract class ForeignKeyIndex<E extends EppResource> extends BackupGroup
    */
   private static <E extends EppResource>
       ImmutableMap<String, ForeignKeyIndex<E>> loadIndexesFromStore(
-          Class<E> clazz, Collection<String> foreignKeys, boolean inTransaction) {
+          Class<E> clazz,
+          Collection<String> foreignKeys,
+          boolean inTransaction,
+          boolean useReplicaJpaTm) {
     if (tm().isOfy()) {
       Class<ForeignKeyIndex<E>> fkiClass = mapToFkiClass(clazz);
       return ImmutableMap.copyOf(
@@ -224,17 +231,18 @@ public abstract class ForeignKeyIndex<E extends EppResource> extends BackupGroup
               : tm().doTransactionless(() -> auditedOfy().load().type(fkiClass).ids(foreignKeys)));
     } else {
       String property = RESOURCE_CLASS_TO_FKI_PROPERTY.get(clazz);
+      JpaTransactionManager jpaTmToUse = useReplicaJpaTm ? replicaJpaTm() : jpaTm();
       ImmutableList<ForeignKeyIndex<E>> indexes =
-          tm().transact(
-                  () ->
-                      jpaTm()
-                          .criteriaQuery(
-                              CriteriaQueryBuilder.create(clazz)
-                                  .whereFieldIsIn(property, foreignKeys)
-                                  .build())
-                          .getResultStream()
-                          .map(e -> ForeignKeyIndex.create(e, e.getDeletionTime()))
-                          .collect(toImmutableList()));
+          jpaTmToUse.transact(
+              () ->
+                  jpaTmToUse
+                      .criteriaQuery(
+                          CriteriaQueryBuilder.create(clazz)
+                              .whereFieldIsIn(property, foreignKeys)
+                              .build())
+                      .getResultStream()
+                      .map(e -> ForeignKeyIndex.create(e, e.getDeletionTime()))
+                      .collect(toImmutableList()));
       // We need to find and return the entities with the maximum deletionTime for each foreign key.
       return Multimaps.index(indexes, ForeignKeyIndex::getForeignKey).asMap().entrySet().stream()
           .map(
@@ -258,7 +266,8 @@ public abstract class ForeignKeyIndex<E extends EppResource> extends BackupGroup
               loadIndexesFromStore(
                       RESOURCE_CLASS_TO_FKI_CLASS.inverse().get(key.getKind()),
                       ImmutableSet.of(foreignKey),
-                      false)
+                      false,
+                      true)
                   .get(foreignKey));
         }
 
@@ -274,7 +283,7 @@ public abstract class ForeignKeyIndex<E extends EppResource> extends BackupGroup
               Streams.stream(keys).map(v -> v.getSqlKey().toString()).collect(toImmutableSet());
           ImmutableSet<VKey<ForeignKeyIndex<?>>> typedKeys = ImmutableSet.copyOf(keys);
           ImmutableMap<String, ? extends ForeignKeyIndex<? extends EppResource>> existingFkis =
-              loadIndexesFromStore(resourceClass, foreignKeys, false);
+              loadIndexesFromStore(resourceClass, foreignKeys, false, true);
           // ofy omits keys that don't have values in Datastore, so re-add them in
           // here with Optional.empty() values.
           return Maps.asMap(
@@ -334,7 +343,7 @@ public abstract class ForeignKeyIndex<E extends EppResource> extends BackupGroup
     // Safe to cast VKey<FKI<E>> to VKey<FKI<?>>
     @SuppressWarnings("unchecked")
     ImmutableList<VKey<ForeignKeyIndex<?>>> fkiVKeys =
-        Streams.stream(foreignKeys)
+        foreignKeys.stream()
             .map(fk -> (VKey<ForeignKeyIndex<?>>) VKey.create(fkiClass, fk))
             .collect(toImmutableList());
     try {

core/src/main/java/google/registry/model/ofy/AugmentedDeleter.java

@@ -23,6 +23,7 @@ import com.googlecode.objectify.Key;
 import com.googlecode.objectify.Result;
 import com.googlecode.objectify.cmd.DeleteType;
 import com.googlecode.objectify.cmd.Deleter;
+import google.registry.model.annotations.DeleteAfterMigration;
 import java.util.Arrays;
 import java.util.stream.Stream;
 
@@ -30,6 +31,7 @@ import java.util.stream.Stream;
  * A Deleter that forwards to {@code auditedOfy().delete()}, but can be augmented via subclassing to
  * do custom processing on the keys to be deleted prior to their deletion.
  */
+@DeleteAfterMigration
 abstract class AugmentedDeleter implements Deleter {
   private final Deleter delegate = ofy().delete();
 

core/src/main/java/google/registry/model/ofy/AugmentedSaver.java

@@ -21,13 +21,15 @@ import com.google.common.collect.ImmutableList;
 import com.googlecode.objectify.Key;
 import com.googlecode.objectify.Result;
 import com.googlecode.objectify.cmd.Saver;
+import google.registry.model.annotations.DeleteAfterMigration;
 import java.util.Arrays;
 import java.util.Map;
 
 /**
- * A Saver that forwards to {@code ofy().save()}, but can be augmented via subclassing to
- * do custom processing on the entities to be saved prior to their saving.
+ * A Saver that forwards to {@code ofy().save()}, but can be augmented via subclassing to do custom
+ * processing on the entities to be saved prior to their saving.
  */
+@DeleteAfterMigration
 abstract class AugmentedSaver implements Saver {
   private final Saver delegate = ofy().save();
 

core/src/main/java/google/registry/model/ofy/CommitLogBucket.java

@@ -31,6 +31,7 @@ import com.googlecode.objectify.annotation.Id;
 import google.registry.config.RegistryConfig;
 import google.registry.model.Buildable;
 import google.registry.model.ImmutableObject;
+import google.registry.model.annotations.DeleteAfterMigration;
 import google.registry.model.annotations.NotBackedUp;
 import google.registry.model.annotations.NotBackedUp.Reason;
 import google.registry.model.replay.DatastoreOnlyEntity;
@@ -51,6 +52,7 @@ import org.joda.time.DateTime;
  */
 @Entity
 @NotBackedUp(reason = Reason.COMMIT_LOGS)
+@DeleteAfterMigration
 public class CommitLogBucket extends ImmutableObject implements Buildable, DatastoreOnlyEntity {
 
   /**

core/src/main/java/google/registry/model/ofy/CommitLogCheckpoint.java

@@ -25,6 +25,7 @@ import com.googlecode.objectify.annotation.Entity;
 import com.googlecode.objectify.annotation.Id;
 import com.googlecode.objectify.annotation.Parent;
 import google.registry.model.ImmutableObject;
+import google.registry.model.annotations.DeleteAfterMigration;
 import google.registry.model.annotations.NotBackedUp;
 import google.registry.model.annotations.NotBackedUp.Reason;
 import google.registry.model.replay.DatastoreOnlyEntity;
@@ -45,6 +46,7 @@ import org.joda.time.DateTime;
  */
 @Entity
 @NotBackedUp(reason = Reason.COMMIT_LOGS)
+@DeleteAfterMigration
 public class CommitLogCheckpoint extends ImmutableObject implements DatastoreOnlyEntity {
 
   /** Shared singleton parent entity for commit log checkpoints. */

core/src/main/java/google/registry/model/ofy/CommitLogCheckpointRoot.java

@@ -21,6 +21,7 @@ import com.googlecode.objectify.Key;
 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 google.registry.model.annotations.NotBackedUp.Reason;
 import google.registry.model.replay.DatastoreOnlyEntity;
@@ -29,6 +30,7 @@ import org.joda.time.DateTime;
 /** Singleton parent entity for all commit log checkpoints. */
 @Entity
 @NotBackedUp(reason = Reason.COMMIT_LOGS)
+@DeleteAfterMigration
 public class CommitLogCheckpointRoot extends ImmutableObject implements DatastoreOnlyEntity {
 
   public static final long SINGLETON_ID = 1;  // There is always exactly one of these.

core/src/main/java/google/registry/model/ofy/CommitLogManifest.java

@@ -23,6 +23,7 @@ import com.googlecode.objectify.annotation.Entity;
 import com.googlecode.objectify.annotation.Id;
 import com.googlecode.objectify.annotation.Parent;
 import google.registry.model.ImmutableObject;
+import google.registry.model.annotations.DeleteAfterMigration;
 import google.registry.model.annotations.NotBackedUp;
 import google.registry.model.annotations.NotBackedUp.Reason;
 import google.registry.model.replay.DatastoreOnlyEntity;
@@ -39,6 +40,7 @@ import org.joda.time.DateTime;
  */
 @Entity
 @NotBackedUp(reason = Reason.COMMIT_LOGS)
+@DeleteAfterMigration
 public class CommitLogManifest extends ImmutableObject implements DatastoreOnlyEntity {
 
   /** Commit log manifests are parented on a random bucket. */

core/src/main/java/google/registry/model/ofy/CommitLogMutation.java

@@ -27,6 +27,7 @@ import com.googlecode.objectify.annotation.Entity;
 import com.googlecode.objectify.annotation.Id;
 import com.googlecode.objectify.annotation.Parent;
 import google.registry.model.ImmutableObject;
+import google.registry.model.annotations.DeleteAfterMigration;
 import google.registry.model.annotations.NotBackedUp;
 import google.registry.model.annotations.NotBackedUp.Reason;
 import google.registry.model.replay.DatastoreOnlyEntity;
@@ -34,6 +35,7 @@ import google.registry.model.replay.DatastoreOnlyEntity;
 /** Representation of a saved entity in a {@link CommitLogManifest} (not deletes). */
 @Entity
 @NotBackedUp(reason = Reason.COMMIT_LOGS)
+@DeleteAfterMigration
 public class CommitLogMutation extends ImmutableObject implements DatastoreOnlyEntity {
 
   /** The manifest this belongs to. */

core/src/main/java/google/registry/model/ofy/CommitLoggedWork.java

@@ -30,6 +30,7 @@ import com.google.common.collect.ImmutableSet;
 import com.googlecode.objectify.Key;
 import google.registry.model.BackupGroupRoot;
 import google.registry.model.ImmutableObject;
+import google.registry.model.annotations.DeleteAfterMigration;
 import google.registry.util.Clock;
 import java.util.HashSet;
 import java.util.Map;
@@ -39,6 +40,7 @@ import java.util.function.Supplier;
 import org.joda.time.DateTime;
 
 /** Wrapper for {@link Supplier} that associates a time with each attempt. */
+@DeleteAfterMigration
 class CommitLoggedWork<R> implements Runnable {
 
   private final Supplier<R> work;

core/src/main/java/google/registry/model/ofy/DatastoreTransactionManager.java

@@ -33,6 +33,7 @@ import com.googlecode.objectify.Key;
 import com.googlecode.objectify.Result;
 import com.googlecode.objectify.cmd.Query;
 import google.registry.model.ImmutableObject;
+import google.registry.model.annotations.DeleteAfterMigration;
 import google.registry.model.annotations.InCrossTld;
 import google.registry.model.contact.ContactHistory;
 import google.registry.model.domain.DomainHistory;
@@ -55,6 +56,7 @@ import javax.persistence.NonUniqueResultException;
 import org.joda.time.DateTime;
 
 /** Datastore implementation of {@link TransactionManager}. */
+@DeleteAfterMigration
 public class DatastoreTransactionManager implements TransactionManager {
 
   private Ofy injectedOfy;

core/src/main/java/google/registry/model/ofy/EntityWritePriorities.java

@@ -16,6 +16,7 @@ package google.registry.model.ofy;
 
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.ImmutableMap;
+import google.registry.model.annotations.DeleteAfterMigration;
 
 /**
  * Contains the mapping from class names to SQL-replay-write priorities.
@@ -26,6 +27,7 @@ import com.google.common.collect.ImmutableMap;
  * values represent an earlier write (and later delete). Higher-valued classes can have foreign keys
  * on lower-valued classes, but not vice versa.
  */
+@DeleteAfterMigration
 public class EntityWritePriorities {
 
   /**

core/src/main/java/google/registry/model/ofy/ObjectifyService.java

@@ -35,6 +35,7 @@ import google.registry.config.RegistryEnvironment;
 import google.registry.model.Buildable;
 import google.registry.model.EntityClasses;
 import google.registry.model.ImmutableObject;
+import google.registry.model.annotations.DeleteAfterMigration;
 import google.registry.model.translators.BloomFilterOfStringTranslatorFactory;
 import google.registry.model.translators.CidrAddressBlockTranslatorFactory;
 import google.registry.model.translators.CommitLogRevisionsTranslatorFactory;
@@ -52,6 +53,7 @@ import google.registry.model.translators.VKeyTranslatorFactory;
  * objects. The class contains a static initializer to call factory().register(...) on all
  * persistable objects in this package.
  */
+@DeleteAfterMigration
 public class ObjectifyService {
 
   /** A singleton instance of our Ofy wrapper. */

core/src/main/java/google/registry/model/ofy/Ofy.java

@@ -37,6 +37,7 @@ import com.googlecode.objectify.ObjectifyFactory;
 import com.googlecode.objectify.cmd.Deleter;
 import com.googlecode.objectify.cmd.Loader;
 import com.googlecode.objectify.cmd.Saver;
+import google.registry.model.annotations.DeleteAfterMigration;
 import google.registry.model.annotations.NotBackedUp;
 import google.registry.model.annotations.VirtualEntity;
 import google.registry.model.ofy.ReadOnlyWork.KillTransactionException;
@@ -59,6 +60,7 @@ import org.joda.time.Duration;
  * simpler to wrap {@link Objectify} rather than extend it because this way we can remove some
  * methods that we don't really want exposed and add some shortcuts.
  */
+@DeleteAfterMigration
 public class Ofy {
 
   private static final FluentLogger logger = FluentLogger.forEnclosingClass();

core/src/main/java/google/registry/model/ofy/OfyFilter.java

@@ -14,6 +14,7 @@
 
 package google.registry.model.ofy;
 
+import google.registry.model.annotations.DeleteAfterMigration;
 import java.io.IOException;
 import javax.servlet.Filter;
 import javax.servlet.FilterChain;
@@ -23,6 +24,7 @@ import javax.servlet.ServletRequest;
 import javax.servlet.ServletResponse;
 
 /** A filter that statically registers types with Objectify. */
+@DeleteAfterMigration
 public class OfyFilter implements Filter {
 
   @Override

core/src/main/java/google/registry/model/ofy/ReadOnlyWork.java

@@ -14,10 +14,12 @@
 
 package google.registry.model.ofy;
 
+import google.registry.model.annotations.DeleteAfterMigration;
 import google.registry.util.Clock;
 import java.util.function.Supplier;
 
 /** Wrapper for {@link Supplier} that disallows mutations and fails the transaction at the end. */
+@DeleteAfterMigration
 class ReadOnlyWork<R> extends CommitLoggedWork<R> {
 
   ReadOnlyWork(Supplier<R> work, Clock clock) {

core/src/main/java/google/registry/model/ofy/ReplayQueue.java

@@ -22,6 +22,7 @@ import com.google.common.collect.ImmutableMap;
 import com.googlecode.objectify.Key;
 import google.registry.config.RegistryEnvironment;
 import google.registry.model.UpdateAutoTimestamp;
+import google.registry.model.annotations.DeleteAfterMigration;
 import google.registry.model.replay.DatastoreEntity;
 import google.registry.model.replay.ReplaySpecializer;
 import google.registry.persistence.VKey;
@@ -34,6 +35,7 @@ import java.util.concurrent.ConcurrentLinkedQueue;
  *
  * <p>This code is to be removed when the actual replay cron job is implemented.
  */
+@DeleteAfterMigration
 public class ReplayQueue {
 
   static ConcurrentLinkedQueue<ImmutableMap<Key<?>, Object>> queue =

core/src/main/java/google/registry/model/ofy/RequestCapturingAsyncDatastoreService.java

@@ -28,6 +28,7 @@ import com.google.appengine.api.datastore.Query;
 import com.google.appengine.api.datastore.Transaction;
 import com.google.appengine.api.datastore.TransactionOptions;
 import com.google.common.collect.ImmutableList;
+import google.registry.model.annotations.DeleteAfterMigration;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
@@ -35,6 +36,7 @@ import java.util.Map;
 import java.util.concurrent.Future;
 
 /** A proxy for {@link AsyncDatastoreService} that exposes call counts. */
+@DeleteAfterMigration
 public class RequestCapturingAsyncDatastoreService implements AsyncDatastoreService {
 
   private final AsyncDatastoreService delegate;

core/src/main/java/google/registry/model/ofy/SessionKeyExposingObjectify.java

@@ -18,8 +18,10 @@ import com.google.common.collect.ImmutableSet;
 import com.googlecode.objectify.Key;
 import com.googlecode.objectify.ObjectifyFactory;
 import com.googlecode.objectify.impl.ObjectifyImpl;
+import google.registry.model.annotations.DeleteAfterMigration;
 
 /** Registry-specific Objectify subclass that exposes the keys used in the current session. */
+@DeleteAfterMigration
 public class SessionKeyExposingObjectify extends ObjectifyImpl<SessionKeyExposingObjectify> {
 
   public SessionKeyExposingObjectify(ObjectifyFactory factory) {

core/src/main/java/google/registry/model/ofy/TimestampInversionException.java

@@ -17,6 +17,7 @@ package google.registry.model.ofy;
 import com.googlecode.objectify.Key;
 import com.googlecode.objectify.Objectify;
 import google.registry.model.BackupGroupRoot;
+import google.registry.model.annotations.DeleteAfterMigration;
 import java.util.Arrays;
 import java.util.Map;
 import org.joda.time.DateTime;
@@ -25,6 +26,7 @@ import org.joda.time.DateTime;
  * Exception when trying to write to Datastore with a timestamp that is inconsistent with a partial
  * ordering on transactions that touch the same entities.
  */
+@DeleteAfterMigration
 class TimestampInversionException extends RuntimeException {
 
   static String getFileAndLine(StackTraceElement callsite) {

core/src/main/java/google/registry/model/ofy/TransactionInfo.java

@@ -26,10 +26,12 @@ import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
 import com.googlecode.objectify.Key;
+import google.registry.model.annotations.DeleteAfterMigration;
 import java.util.Map;
 import org.joda.time.DateTime;
 
 /** Metadata for an {@link Ofy} transaction that saves commit logs. */
+@DeleteAfterMigration
 public class TransactionInfo {
 
   @VisibleForTesting

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

@@ -14,9 +14,11 @@
 
 package google.registry.model.replay;
 
+import google.registry.model.annotations.DeleteAfterMigration;
 import java.util.Optional;
 
 /** An entity that has the same Java object representation in SQL and Datastore. */
+@DeleteAfterMigration
 public interface DatastoreAndSqlEntity extends DatastoreEntity, SqlEntity {
 
   @Override

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

@@ -14,6 +14,7 @@
 
 package google.registry.model.replay;
 
+import google.registry.model.annotations.DeleteAfterMigration;
 import java.util.Optional;
 
 /**
@@ -24,6 +25,7 @@ import java.util.Optional;
  * transactions and data into the secondary SQL store during the first, Datastore-primary, phase of
  * the migration.
  */
+@DeleteAfterMigration
 public interface DatastoreEntity {
 
   Optional<SqlEntity> toSqlEntity();

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

@@ -14,9 +14,11 @@
 
 package google.registry.model.replay;
 
+import google.registry.model.annotations.DeleteAfterMigration;
 import java.util.Optional;
 
 /** An entity that is only stored in Datastore, that should not be replayed to SQL. */
+@DeleteAfterMigration
 public interface DatastoreOnlyEntity extends DatastoreEntity {
   @Override
   default Optional<SqlEntity> toSqlEntity() {

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

@@ -22,9 +22,11 @@ import com.googlecode.objectify.Key;
 import com.googlecode.objectify.annotation.Entity;
 import com.googlecode.objectify.annotation.Id;
 import google.registry.model.ImmutableObject;
+import google.registry.model.annotations.DeleteAfterMigration;
 
 /** Datastore entity to keep track of the last SQL transaction imported into the datastore. */
 @Entity
+@DeleteAfterMigration
 public class LastSqlTransaction extends ImmutableObject implements DatastoreOnlyEntity {
 
   /** The key for this singleton. */

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

@@ -14,6 +14,7 @@
 
 package google.registry.model.replay;
 
+import google.registry.model.annotations.DeleteAfterMigration;
 import java.util.Optional;
 
 /**
@@ -21,6 +22,7 @@ import java.util.Optional;
  *
  * <p>We expect that this is a result of the entity being dually-written.
  */
+@DeleteAfterMigration
 public interface NonReplicatedEntity extends DatastoreEntity, SqlEntity {
 
   @Override

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

@@ -14,6 +14,7 @@
 
 package google.registry.model.replay;
 
+import google.registry.model.annotations.DeleteAfterMigration;
 import google.registry.persistence.VKey;
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
@@ -25,6 +26,7 @@ import java.lang.reflect.Method;
  * not directly present in the other database. This class allows us to do that by using reflection
  * to invoke special class methods if they are present.
  */
+@DeleteAfterMigration
 public class ReplaySpecializer {
 
   public static void beforeSqlDelete(VKey<?> key) {

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

@@ -27,6 +27,7 @@ import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.ImmutableList;
 import com.google.common.flogger.FluentLogger;
 import google.registry.model.UpdateAutoTimestamp;
+import google.registry.model.annotations.DeleteAfterMigration;
 import google.registry.model.common.DatabaseMigrationStateSchedule;
 import google.registry.model.common.DatabaseMigrationStateSchedule.MigrationState;
 import google.registry.model.common.DatabaseMigrationStateSchedule.ReplayDirection;
@@ -53,10 +54,15 @@ import org.joda.time.Duration;
     automaticallyPrintOk = true,
     auth = Auth.AUTH_INTERNAL_OR_ADMIN)
 @VisibleForTesting
+@DeleteAfterMigration
 public class ReplicateToDatastoreAction implements Runnable {
   public static final String PATH = "/_dr/cron/replicateToDatastore";
   private static final FluentLogger logger = FluentLogger.forEnclosingClass();
 
+  /** Name of the lock that ensures sequential execution of replays. */
+  public static final String REPLICATE_TO_DATASTORE_LOCK_NAME =
+      ReplicateToDatastoreAction.class.getSimpleName();
+
   /**
    * Number of transactions to fetch from SQL. The rationale for 200 is that we're processing these
    * every minute and our production instance currently does about 2 mutations per second, so this
@@ -64,7 +70,7 @@ public class ReplicateToDatastoreAction implements Runnable {
    */
   public static final int BATCH_SIZE = 200;
 
-  private static final Duration LEASE_LENGTH = standardHours(1);
+  public static final Duration REPLICATE_TO_DATASTORE_LOCK_LEASE_LENGTH = standardHours(1);
 
   private final Clock clock;
   private final RequestStatusChecker requestStatusChecker;
@@ -79,21 +85,26 @@ public class ReplicateToDatastoreAction implements Runnable {
   }
 
   @VisibleForTesting
-  public List<TransactionEntity> getTransactionBatch() {
+  public List<TransactionEntity> getTransactionBatchAtSnapshot() {
+    return getTransactionBatchAtSnapshot(Optional.empty());
+  }
+
+  static List<TransactionEntity> getTransactionBatchAtSnapshot(Optional<String> snapshotId) {
     // Get the next batch of transactions that we haven't replicated.
     LastSqlTransaction lastSqlTxnBeforeBatch = ofyTm().transact(LastSqlTransaction::load);
     try {
       return jpaTm()
           .transactWithoutBackup(
-              () ->
-                  jpaTm()
-                      .query(
-                          "SELECT txn FROM TransactionEntity txn WHERE id >"
-                              + " :lastId ORDER BY id",
-                          TransactionEntity.class)
-                      .setParameter("lastId", lastSqlTxnBeforeBatch.getTransactionId())
-                      .setMaxResults(BATCH_SIZE)
-                      .getResultList());
+              () -> {
+                snapshotId.ifPresent(jpaTm()::setDatabaseSnapshot);
+                return jpaTm()
+                    .query(
+                        "SELECT txn FROM TransactionEntity txn WHERE id >" + " :lastId ORDER BY id",
+                        TransactionEntity.class)
+                    .setParameter("lastId", lastSqlTxnBeforeBatch.getTransactionId())
+                    .setMaxResults(BATCH_SIZE)
+                    .getResultList();
+              });
     } catch (NoResultException e) {
       return ImmutableList.of();
     }
@@ -106,7 +117,7 @@ public class ReplicateToDatastoreAction implements Runnable {
    * <p>Throws an exception if a fatal error occurred and the batch should be aborted
    */
   @VisibleForTesting
-  public void applyTransaction(TransactionEntity txnEntity) {
+  public static void applyTransaction(TransactionEntity txnEntity) {
     logger.atInfo().log("Applying a single transaction Cloud SQL -> Cloud Datastore.");
     try (UpdateAutoTimestamp.DisableAutoUpdateResource disabler =
         UpdateAutoTimestamp.disableAutoUpdate()) {
@@ -172,7 +183,11 @@ public class ReplicateToDatastoreAction implements Runnable {
     }
     Optional<Lock> lock =
         Lock.acquireSql(
-            this.getClass().getSimpleName(), null, LEASE_LENGTH, requestStatusChecker, false);
+            REPLICATE_TO_DATASTORE_LOCK_NAME,
+            null,
+            REPLICATE_TO_DATASTORE_LOCK_LEASE_LENGTH,
+            requestStatusChecker,
+            false);
     if (!lock.isPresent()) {
       String message = "Can't acquire ReplicateToDatastoreAction lock, aborting.";
       logger.atSevere().log(message);
@@ -201,10 +216,14 @@ public class ReplicateToDatastoreAction implements Runnable {
   }
 
   private int replayAllTransactions() {
+    return replayAllTransactions(Optional.empty());
+  }
+
+  public static int replayAllTransactions(Optional<String> snapshotId) {
     int numTransactionsReplayed = 0;
     List<TransactionEntity> transactionBatch;
     do {
-      transactionBatch = getTransactionBatch();
+      transactionBatch = getTransactionBatchAtSnapshot(snapshotId);
       for (TransactionEntity transaction : transactionBatch) {
         applyTransaction(transaction);
         numTransactionsReplayed++;

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

@@ -16,6 +16,7 @@ package google.registry.model.replay;
 
 import static google.registry.persistence.transaction.TransactionManagerFactory.jpaTm;
 
+import google.registry.model.annotations.DeleteAfterMigration;
 import java.util.Optional;
 
 /**
@@ -25,6 +26,7 @@ import java.util.Optional;
  * <p>This will be used when replaying SQL transactions into Datastore, during the second,
  * SQL-primary, phase of the migration from Datastore to SQL.
  */
+@DeleteAfterMigration
 public interface SqlEntity {
 
   Optional<DatastoreEntity> toDatastoreEntity();

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

@@ -14,9 +14,11 @@
 
 package google.registry.model.replay;
 
+import google.registry.model.annotations.DeleteAfterMigration;
 import java.util.Optional;
 
 /** An entity that is only stored in SQL, that should not be replayed to Datastore. */
+@DeleteAfterMigration
 public interface SqlOnlyEntity extends SqlEntity {
   @Override
   default Optional<DatastoreEntity> toDatastoreEntity() {

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

@@ -17,12 +17,14 @@ package google.registry.model.replay;
 import static google.registry.persistence.transaction.TransactionManagerFactory.jpaTm;
 import static google.registry.util.DateTimeUtils.START_OF_TIME;
 
+import google.registry.model.annotations.DeleteAfterMigration;
 import google.registry.model.common.CrossTldSingleton;
 import javax.persistence.Column;
 import javax.persistence.Entity;
 import org.joda.time.DateTime;
 
 @Entity
+@DeleteAfterMigration
 public class SqlReplayCheckpoint extends CrossTldSingleton implements SqlOnlyEntity {
 
   @Column(nullable = false)