Add new domain list fields to Registry objects (#1208)

* Add domain list name fields to Registry objects

* Add some comments

* Added scrap command

* Fix typo

* capitalize TLD

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

@@ -66,6 +66,8 @@ import org.joda.time.Duration;
     service = Action.Service.BACKEND,
     path = "/_dr/task/deleteOldCommitLogs",
     auth = Auth.AUTH_INTERNAL_OR_ADMIN)
+// No longer needed in SQL. Subject to future removal.
+@Deprecated
 public final class DeleteOldCommitLogsAction implements Runnable {
 
   private static final int NUM_MAP_SHARDS = 20;

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

@@ -34,6 +34,7 @@ import google.registry.model.server.Lock;
 import google.registry.model.translators.VKeyTranslatorFactory;
 import google.registry.persistence.VKey;
 import google.registry.request.Action;
+import google.registry.request.Action.Method;
 import google.registry.request.Response;
 import google.registry.request.auth.Auth;
 import google.registry.schema.replay.DatastoreEntity;
@@ -56,7 +57,7 @@ import org.joda.time.Duration;
 @Action(
     service = Action.Service.BACKEND,
     path = ReplayCommitLogsToSqlAction.PATH,
-    method = Action.Method.POST,
+    method = Method.POST,
     automaticallyPrintOk = true,
     auth = Auth.AUTH_INTERNAL_OR_ADMIN)
 public class ReplayCommitLogsToSqlAction implements Runnable {

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

@@ -127,10 +127,11 @@ public class ExpandRecurringBillingEventsAction implements Runnable {
                       jpaTm()
                           .query(
                               "FROM BillingRecurrence "
-                                  + "WHERE event_time <= :executeTime "
-                                  + "AND event_time < recurrence_end_time",
+                                  + "WHERE eventTime <= :executeTime "
+                                  + "AND eventTime < recurrenceEndTime "
+                                  + "ORDER BY id ASC",
                               Recurring.class)
-                          .setParameter("executeTime", executeTime.toDate())
+                          .setParameter("executeTime", executeTime)
                           // Need to get a list from the transaction and then convert it to a stream
                           // for further processing. If we get a stream directly, each elements gets
                           // processed downstream eagerly but Hibernate returns a

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

@@ -54,6 +54,8 @@ import javax.inject.Inject;
     service = Action.Service.BACKEND,
     path = "/_dr/task/resaveAllEppResources",
     auth = Auth.AUTH_INTERNAL_OR_ADMIN)
+// No longer needed in SQL. Subject to future removal.
+@Deprecated
 public class ResaveAllEppResourcesAction implements Runnable {
 
   @Inject MapreduceRunner mrRunner;

core/src/main/java/google/registry/beam/spec11/Spec11Pipeline.java

@@ -23,8 +23,10 @@ import dagger.Component;
 import dagger.Module;
 import dagger.Provides;
 import google.registry.beam.common.RegistryJpaIO;
+import google.registry.beam.common.RegistryJpaIO.Read;
 import google.registry.beam.spec11.SafeBrowsingTransforms.EvaluateSafeBrowsingFn;
 import google.registry.config.RegistryConfig.ConfigModule;
+import google.registry.model.domain.DomainBase;
 import google.registry.model.reporting.Spec11ThreatMatch;
 import google.registry.model.reporting.Spec11ThreatMatch.ThreatType;
 import google.registry.util.Retrier;
@@ -97,20 +99,9 @@ public class Spec11Pipeline implements Serializable {
 
   void setupPipeline(Pipeline pipeline) {
     PCollection<Subdomain> domains =
-        pipeline.apply(
-            "Read active domains from BigQuery",
-            BigQueryIO.read(Subdomain::parseFromRecord)
-                .fromQuery(
-                    SqlTemplate.create(getQueryFromFile(Spec11Pipeline.class, "subdomains.sql"))
-                        .put("PROJECT_ID", options.getProject())
-                        .put("DATASTORE_EXPORT_DATASET", "latest_datastore_export")
-                        .put("REGISTRAR_TABLE", "Registrar")
-                        .put("DOMAIN_BASE_TABLE", "DomainBase")
-                        .build())
-                .withCoder(SerializableCoder.of(Subdomain.class))
-                .usingStandardSql()
-                .withoutValidation()
-                .withTemplateCompatibility());
+        options.getDatabase().equals("DATASTORE")
+            ? readFromBigQuery(options, pipeline)
+            : readFromCloudSql(pipeline);
 
     PCollection<KV<Subdomain, ThreatMatch>> threatMatches =
         domains.apply("Run through SafeBrowsing API", ParDo.of(safeBrowsingFn));
@@ -119,6 +110,47 @@ public class Spec11Pipeline implements Serializable {
     saveToGcs(threatMatches, options);
   }
 
+  static PCollection<Subdomain> readFromCloudSql(Pipeline pipeline) {
+    Read<Object[], Subdomain> read =
+        RegistryJpaIO.read(
+            "select d, r.emailAddress from Domain d join Registrar r on"
+                + " d.currentSponsorClientId = r.clientIdentifier where r.type = 'REAL'"
+                + " and d.deletionTime > now()",
+            Spec11Pipeline::parseRow);
+
+    return pipeline.apply("Read active domains from Cloud SQL", read);
+  }
+
+  static PCollection<Subdomain> readFromBigQuery(Spec11PipelineOptions options, Pipeline pipeline) {
+    return pipeline.apply(
+        "Read active domains from BigQuery",
+        BigQueryIO.read(Subdomain::parseFromRecord)
+            .fromQuery(
+                SqlTemplate.create(getQueryFromFile(Spec11Pipeline.class, "subdomains.sql"))
+                    .put("PROJECT_ID", options.getProject())
+                    .put("DATASTORE_EXPORT_DATASET", "latest_datastore_export")
+                    .put("REGISTRAR_TABLE", "Registrar")
+                    .put("DOMAIN_BASE_TABLE", "DomainBase")
+                    .build())
+            .withCoder(SerializableCoder.of(Subdomain.class))
+            .usingStandardSql()
+            .withoutValidation()
+            .withTemplateCompatibility());
+  }
+
+  private static Subdomain parseRow(Object[] row) {
+    DomainBase domainBase = (DomainBase) row[0];
+    String emailAddress = (String) row[1];
+    if (emailAddress == null) {
+      emailAddress = "";
+    }
+    return Subdomain.create(
+        domainBase.getDomainName(),
+        domainBase.getRepoId(),
+        domainBase.getCurrentSponsorClientId(),
+        emailAddress);
+  }
+
   static void saveToSql(
       PCollection<KV<Subdomain, ThreatMatch>> threatMatches, Spec11PipelineOptions options) {
     String transformId = "Spec11 Threat Matches";

core/src/main/java/google/registry/beam/spec11/Spec11PipelineOptions.java

@@ -34,4 +34,9 @@ public interface Spec11PipelineOptions extends RegistryPipelineOptions {
   String getReportingBucketUrl();
 
   void setReportingBucketUrl(String value);
+
+  @Description("The database to read data from.")
+  String getDatabase();
+
+  void setDatabase(String value);
 }

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

@@ -200,7 +200,7 @@
   </cron>
 
   <cron>
-    <url><![CDATA[/_dr/task/replayCommitLogsToSql]]></url>
+    <url><![CDATA[/_dr/cron/fanout?queue=replay-commit-logs-to-sql&endpoint=/_dr/task/replayCommitLogsToSql&runInEmpty]]></url>
     <description>
       Replays recent commit logs from Datastore to the SQL secondary backend.
     </description>

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

@@ -237,6 +237,12 @@
     <url-pattern>/_dr/task/killCommitLogs</url-pattern>
   </servlet-mapping>
 
+  <!-- Replays Datastore commit logs to SQL. -->
+  <servlet-mapping>
+    <servlet-name>backend-servlet</servlet-name>
+    <url-pattern>/_dr/task/replayCommitLogsToSql</url-pattern>
+  </servlet-mapping>
+
   <!-- MapReduce servlet. -->
   <servlet>
     <servlet-name>mapreduce</servlet-name>

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

@@ -201,7 +201,7 @@
   </cron>
 
   <cron>
-    <url><![CDATA[/_dr/task/replayCommitLogsToSql]]></url>
+    <url><![CDATA[/_dr/cron/fanout?queue=replay-commit-logs-to-sql&endpoint=/_dr/task/replayCommitLogsToSql&runInEmpty]]></url>
     <description>
       Replays recent commit logs from Datastore to the SQL secondary backend.
     </description>

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

@@ -1,6 +1,17 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <cronentries>
 
+  <cron>
+    <url>/_dr/task/rdeStaging</url>
+    <description>
+      This job generates a full RDE escrow deposit as a single gigantic XML document
+      and streams it to cloud storage. When this job has finished successfully, it'll
+      launch a separate task that uploads the deposit file to Iron Mountain via SFTP.
+    </description>
+    <schedule>every day 00:07</schedule>
+    <target>backend</target>
+  </cron>
+
   <cron>
     <url><![CDATA[/_dr/cron/readDnsQueue?jitterSeconds=45]]></url>
     <description>
@@ -101,11 +112,20 @@
   </cron>
 
   <cron>
-    <url><![CDATA[/_dr/task/replayCommitLogsToSql]]></url>
+    <url><![CDATA[/_dr/cron/fanout?queue=replay-commit-logs-to-sql&endpoint=/_dr/task/replayCommitLogsToSql&runInEmpty]]></url>
     <description>
       Replays recent commit logs from Datastore to the SQL secondary backend.
     </description>
     <schedule>every 3 minutes</schedule>
     <target>backend</target>
   </cron>
+
+  <cron>
+    <url><![CDATA[/_dr/cron/commitLogCheckpoint]]></url>
+    <description>
+      This job checkpoints the commit log buckets and exports the diff since last checkpoint to GCS.
+    </description>
+    <schedule>every 3 minutes synchronized</schedule>
+    <target>backend</target>
+  </cron>
 </cronentries>

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

@@ -230,7 +230,7 @@
   </cron>
 
   <cron>
-    <url><![CDATA[/_dr/task/replayCommitLogsToSql]]></url>
+    <url><![CDATA[/_dr/cron/fanout?queue=replay-commit-logs-to-sql&endpoint=/_dr/task/replayCommitLogsToSql&runInEmpty]]></url>
     <description>
       Replays recent commit logs from Datastore to the SQL secondary backend.
     </description>

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

@@ -19,9 +19,10 @@ import static com.google.common.base.Verify.verifyNotNull;
 import static google.registry.mapreduce.inputs.EppResourceInputs.createEntityInput;
 import static google.registry.model.EppResourceUtils.isActive;
 import static google.registry.model.registry.Registries.getTldsOfType;
+import static google.registry.persistence.transaction.TransactionManagerFactory.jpaTm;
+import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
 import static google.registry.request.Action.Method.POST;
 import static java.nio.charset.StandardCharsets.UTF_8;
-import static org.joda.time.DateTimeZone.UTC;
 
 import com.google.appengine.tools.cloudstorage.GcsFilename;
 import com.google.appengine.tools.cloudstorage.RetryParams;
@@ -45,12 +46,14 @@ import google.registry.request.Action;
 import google.registry.request.Response;
 import google.registry.request.auth.Auth;
 import google.registry.storage.drive.DriveConnection;
+import google.registry.util.Clock;
 import google.registry.util.NonFinalForTesting;
 import java.io.IOException;
 import java.io.ObjectInputStream;
 import java.io.OutputStream;
 import java.io.OutputStreamWriter;
 import java.io.Writer;
+import java.util.List;
 import java.util.function.Supplier;
 import javax.inject.Inject;
 import org.joda.time.DateTime;
@@ -70,9 +73,13 @@ public class ExportDomainListsAction implements Runnable {
 
   private static final FluentLogger logger = FluentLogger.forEnclosingClass();
   private static final int MAX_NUM_REDUCE_SHARDS = 100;
+  public static final String REGISTERED_DOMAINS_FILENAME = "registered_domains.txt";
 
   @Inject MapreduceRunner mrRunner;
   @Inject Response response;
+  @Inject Clock clock;
+  @Inject DriveConnection driveConnection;
+
   @Inject @Config("domainListsGcsBucket") String gcsBucket;
   @Inject @Config("gcsBufferSize") int gcsBufferSize;
   @Inject ExportDomainListsAction() {}
@@ -81,15 +88,99 @@ public class ExportDomainListsAction implements Runnable {
   public void run() {
     ImmutableSet<String> realTlds = getTldsOfType(TldType.REAL);
     logger.atInfo().log("Exporting domain lists for tlds %s", realTlds);
-    mrRunner
-        .setJobName("Export domain lists")
-        .setModuleName("backend")
-        .setDefaultReduceShards(Math.min(realTlds.size(), MAX_NUM_REDUCE_SHARDS))
-        .runMapreduce(
-            new ExportDomainListsMapper(DateTime.now(UTC), realTlds),
-            new ExportDomainListsReducer(gcsBucket, gcsBufferSize),
-            ImmutableList.of(createEntityInput(DomainBase.class)))
-        .sendLinkToMapreduceConsole(response);
+    if (tm().isOfy()) {
+      mrRunner
+          .setJobName("Export domain lists")
+          .setModuleName("backend")
+          .setDefaultReduceShards(Math.min(realTlds.size(), MAX_NUM_REDUCE_SHARDS))
+          .runMapreduce(
+              new ExportDomainListsMapper(clock.nowUtc(), realTlds),
+              new ExportDomainListsReducer(gcsBucket, gcsBufferSize),
+              ImmutableList.of(createEntityInput(DomainBase.class)))
+          .sendLinkToMapreduceConsole(response);
+    } else {
+      realTlds.forEach(
+          tld -> {
+            List<String> domains =
+                tm().transact(
+                        () ->
+                            // Note that if we had "creationTime <= :now" in the condition (not
+                            // necessary as there is no pending creation, the order of deletionTime
+                            // and creationTime in the query would have been significant and it
+                            // should come after deletionTime. When Hibernate substitutes "now" it
+                            // will first validate that the **first** field that is to be compared
+                            // with it (deletionTime) is assignable from the substituted Java object
+                            // (click.nowUtc()). Since creationTime is a CreateAutoTimestamp, if it
+                            // comes first, we will need to substitute "now" with
+                            // CreateAutoTimestamp.create(clock.nowUtc()). This might look a bit
+                            // strange as the Java object type is clearly incompatible between the
+                            // two fields deletionTime (DateTime) and creationTime, yet they are
+                            // compared with the same "now". It is actually OK because in the end
+                            // Hibernate converts everything to SQL types (and Java field names to
+                            // SQL column names) to run the query. Both CreateAutoTimestamp and
+                            // DateTime are persisted as timestamp_z in SQL. It is only the
+                            // validation that compares the Java types, and only with the first
+                            // field that compares with the substituted value.
+                            jpaTm()
+                                .query(
+                                    "SELECT fullyQualifiedDomainName FROM Domain "
+                                        + "WHERE tld = :tld "
+                                        + "AND deletionTime > :now "
+                                        + "ORDER by fullyQualifiedDomainName ASC",
+                                    String.class)
+                                .setParameter("tld", tld)
+                                .setParameter("now", clock.nowUtc())
+                                .getResultList());
+            String domainsList = Joiner.on("\n").join(domains);
+            logger.atInfo().log(
+                "Exporting %d domains for TLD %s to GCS and Drive.", domains.size(), tld);
+            exportToGcs(tld, domainsList, gcsBucket, gcsBufferSize);
+            exportToDrive(tld, domainsList, driveConnection);
+          });
+    }
+  }
+
+  protected static boolean exportToDrive(
+      String tld, String domains, DriveConnection driveConnection) {
+    verifyNotNull(driveConnection, "Expecting non-null driveConnection");
+    try {
+      Registry registry = Registry.get(tld);
+      if (registry.getDriveFolderId() == null) {
+        logger.atInfo().log(
+            "Skipping registered domains export for TLD %s because Drive folder isn't specified",
+            tld);
+      } else {
+        String resultMsg =
+            driveConnection.createOrUpdateFile(
+                REGISTERED_DOMAINS_FILENAME,
+                MediaType.PLAIN_TEXT_UTF_8,
+                registry.getDriveFolderId(),
+                domains.getBytes(UTF_8));
+        logger.atInfo().log(
+            "Exporting registered domains succeeded for TLD %s, response was: %s", tld, resultMsg);
+      }
+    } catch (Throwable e) {
+      logger.atSevere().withCause(e).log(
+          "Error exporting registered domains for TLD %s to Drive, skipping...", tld);
+      return false;
+    }
+    return true;
+  }
+
+  protected static boolean exportToGcs(
+      String tld, String domains, String gcsBucket, int gcsBufferSize) {
+    GcsFilename filename = new GcsFilename(gcsBucket, tld + ".txt");
+    GcsUtils cloudStorage =
+        new GcsUtils(createGcsService(RetryParams.getDefaultInstance()), gcsBufferSize);
+    try (OutputStream gcsOutput = cloudStorage.openOutputStream(filename);
+        Writer osWriter = new OutputStreamWriter(gcsOutput, UTF_8)) {
+      osWriter.write(domains);
+    } catch (Throwable e) {
+      logger.atSevere().withCause(e).log(
+          "Error exporting registered domains for TLD %s to GCS, skipping...", tld);
+      return false;
+    }
+    return true;
   }
 
   static class ExportDomainListsMapper extends Mapper<DomainBase, String, String> {
@@ -122,9 +213,6 @@ public class ExportDomainListsAction implements Runnable {
     private static Supplier<DriveConnection> driveConnectionSupplier =
         Suppliers.memoize(() -> DaggerDriveModule_DriveComponent.create().driveConnection());
 
-    static final String REGISTERED_DOMAINS_FILENAME = "registered_domains.txt";
-    static final MediaType EXPORT_MIME_TYPE = MediaType.PLAIN_TEXT_UTF_8;
-
     private final String gcsBucket;
     private final int gcsBufferSize;
 
@@ -147,53 +235,21 @@ public class ExportDomainListsAction implements Runnable {
       driveConnection = driveConnectionSupplier.get();
     }
 
-    private void exportToDrive(String tld, String domains) {
-      verifyNotNull(driveConnection, "expecting non-null driveConnection");
-      try {
-        Registry registry = Registry.get(tld);
-        if (registry.getDriveFolderId() == null) {
-          logger.atInfo().log(
-              "Skipping registered domains export for TLD %s because Drive folder isn't specified",
-              tld);
-        } else {
-          String resultMsg =
-              driveConnection.createOrUpdateFile(
-                  REGISTERED_DOMAINS_FILENAME,
-                  EXPORT_MIME_TYPE,
-                  registry.getDriveFolderId(),
-                  domains.getBytes(UTF_8));
-          logger.atInfo().log(
-              "Exporting registered domains succeeded for TLD %s, response was: %s",
-              tld, resultMsg);
-        }
-      } catch (Throwable e) {
-        logger.atSevere().withCause(e).log(
-            "Error exporting registered domains for TLD %s to Drive", tld);
-      }
-      getContext().incrementCounter("domain lists written out to Drive");
-    }
-
-    private void exportToGcs(String tld, String domains) {
-      GcsFilename filename = new GcsFilename(gcsBucket, tld + ".txt");
-      GcsUtils cloudStorage =
-          new GcsUtils(createGcsService(RetryParams.getDefaultInstance()), gcsBufferSize);
-      try (OutputStream gcsOutput = cloudStorage.openOutputStream(filename);
-          Writer osWriter = new OutputStreamWriter(gcsOutput, UTF_8)) {
-        osWriter.write(domains);
-      } catch (IOException e) {
-        logger.atSevere().withCause(e).log(
-            "Error exporting registered domains for TLD %s to GCS.", tld);
-      }
-      getContext().incrementCounter("domain lists written out to GCS");
-    }
-
     @Override
     public void reduce(String tld, ReducerInput<String> fqdns) {
       ImmutableList<String> domains = ImmutableList.sortedCopyOf(() -> fqdns);
       String domainsList = Joiner.on('\n').join(domains);
       logger.atInfo().log("Exporting %d domains for TLD %s to GCS and Drive.", domains.size(), tld);
-      exportToGcs(tld, domainsList);
-      exportToDrive(tld, domainsList);
+      if (exportToGcs(tld, domainsList, gcsBucket, gcsBufferSize)) {
+        getContext().incrementCounter("domain lists successful written out to GCS");
+      } else {
+        getContext().incrementCounter("domain lists failed to write out to GCS");
+      }
+      if (exportToDrive(tld, domainsList, driveConnection)) {
+        getContext().incrementCounter("domain lists successfully written out to Drive");
+      } else {
+        getContext().incrementCounter("domain lists failed to write out to Drive");
+      }
     }
 
     @VisibleForTesting

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

@@ -17,10 +17,10 @@ package google.registry.model;
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.collect.ImmutableSet.toImmutableSet;
 import static google.registry.model.ofy.ObjectifyService.auditedOfy;
-import static google.registry.model.ofy.ObjectifyService.ofy;
 import static google.registry.persistence.transaction.TransactionManagerFactory.jpaTm;
 import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
 import static google.registry.persistence.transaction.TransactionManagerUtil.transactIfJpaTm;
+import static google.registry.util.DateTimeUtils.START_OF_TIME;
 import static google.registry.util.DateTimeUtils.isAtOrAfter;
 import static google.registry.util.DateTimeUtils.isBeforeOrAt;
 import static google.registry.util.DateTimeUtils.latestOf;
@@ -29,8 +29,6 @@ import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableSet;
 import com.google.common.flogger.FluentLogger;
 import com.googlecode.objectify.Key;
-import com.googlecode.objectify.Result;
-import com.googlecode.objectify.util.ResultNow;
 import google.registry.config.RegistryConfig;
 import google.registry.model.EppResource.BuilderWithTransferData;
 import google.registry.model.EppResource.ForeignKeyedEppResource;
@@ -43,14 +41,18 @@ import google.registry.model.index.ForeignKeyIndex;
 import google.registry.model.ofy.CommitLogManifest;
 import google.registry.model.ofy.CommitLogMutation;
 import google.registry.model.registry.Registry;
+import google.registry.model.reporting.HistoryEntry;
+import google.registry.model.reporting.HistoryEntryDao;
 import google.registry.model.transfer.DomainTransferData;
 import google.registry.model.transfer.TransferData;
 import google.registry.model.transfer.TransferStatus;
 import google.registry.persistence.VKey;
+import java.util.Comparator;
 import java.util.List;
 import java.util.Map.Entry;
 import java.util.Optional;
 import java.util.function.Function;
+import java.util.function.Supplier;
 import javax.annotation.Nullable;
 import javax.persistence.Query;
 import org.joda.time.DateTime;
@@ -266,80 +268,147 @@ public final class EppResourceUtils {
    * Rewinds an {@link EppResource} object to a given point in time.
    *
    * <p>This method costs nothing if {@code resource} is already current. Otherwise it needs to
-   * perform a single asynchronous key fetch operation.
+   * perform a single fetch operation.
    *
    * <p><b>Warning:</b> A resource can only be rolled backwards in time, not forwards; therefore
    * {@code resource} should be whatever's currently in Datastore.
    *
-   * <p><b>Warning:</b> Revisions are granular to 24-hour periods. It's recommended that
-   * {@code timestamp} be set to midnight. Otherwise you must take into consideration that under
-   * certain circumstances, a resource might be restored to a revision on the previous day, even if
-   * there were revisions made earlier on the same date as {@code timestamp}; however, a resource
-   * will never be restored to a revision occurring after {@code timestamp}. This behavior is due to
-   * the way {@link google.registry.model.translators.CommitLogRevisionsTranslatorFactory
+   * <p><b>Warning:</b> In Datastore, revisions are granular to 24-hour periods. It's recommended
+   * that {@code timestamp} be set to midnight. If you don't use midnight, you must take into
+   * consideration that under certain circumstances, a resource might be restored to a revision on
+   * the previous day, even if there were revisions made earlier on the same date as {@code
+   * timestamp}; however, a resource will never be restored to a revision occurring after {@code
+   * timestamp}. This behavior is due to the way {@link
+   * google.registry.model.translators.CommitLogRevisionsTranslatorFactory
    * CommitLogRevisionsTranslatorFactory} manages the {@link EppResource#revisions} field. Please
    * note however that the creation and deletion times of a resource are granular to the
    * millisecond.
    *
-   * @return an asynchronous operation returning resource at {@code timestamp} or {@code null} if
-   *     resource is deleted or not yet created
+   * <p>Example: a resource in Datastore has three revisions A, B, and C
+   *
+   * <ul>
+   *   <li>A: Day 0, 1pm
+   *   <li>B: Day 1, 1pm
+   *   <li>C: Day 1, 3pm
+   * </ul>
+   *
+   * <p>If one requests the resource as of day 1 at 2pm, we will return revision A because as far as
+   * the commit logs are concerned, revision C completely overwrites the existence of revision B.
+   *
+   * <p>When using the SQL backend (post-Registry-3.0-migration) this restriction goes away and
+   * objects can be restored to any revision.
+   *
+   * @return the resource at {@code timestamp} or {@code null} if resource is deleted or not yet
+   *     created
    */
-  public static <T extends EppResource>
-      Result<T> loadAtPointInTime(final T resource, final DateTime timestamp) {
+  public static <T extends EppResource> T loadAtPointInTime(
+      final T resource, final DateTime timestamp) {
     // If we're before the resource creation time, don't try to find a "most recent revision".
     if (timestamp.isBefore(resource.getCreationTime())) {
-      return new ResultNow<>(null);
+      return null;
     }
     // If the resource was not modified after the requested time, then use it as-is, otherwise find
-    // the most recent revision asynchronously, and return an async result that wraps that revision
-    // and returns it projected forward to exactly the desired timestamp, or null if the resource is
-    // deleted at that timestamp.
-    final Result<T> loadResult =
+    // the most recent revision and project it forward to exactly the desired timestamp, or null if
+    // the resource is deleted at that timestamp.
+    T loadedResource =
         isAtOrAfter(timestamp, resource.getUpdateTimestamp().getTimestamp())
-            ? new ResultNow<>(resource)
+            ? resource
             : loadMostRecentRevisionAtTime(resource, timestamp);
-    return () -> {
-      T loadedResource = loadResult.now();
-      return (loadedResource == null) ? null
-          : (isActive(loadedResource, timestamp)
-              ? cloneProjectedAtTime(loadedResource, timestamp)
-              : null);
-    };
+    return (loadedResource == null)
+        ? null
+        : (isActive(loadedResource, timestamp)
+            ? cloneProjectedAtTime(loadedResource, timestamp)
+            : null);
   }
 
   /**
-   * Returns an asynchronous result holding the most recent Datastore revision of a given
-   * EppResource before or at the provided timestamp using the EppResource revisions map, falling
-   * back to using the earliest revision or the resource as-is if there are no revisions.
+   * Rewinds an {@link EppResource} object to a given point in time.
+   *
+   * <p>This method costs nothing if {@code resource} is already current. Otherwise it returns an
+   * async operation that performs a single fetch operation.
+   *
+   * @return an asynchronous operation returning resource at {@code timestamp} or {@code null} if
+   *     resource is deleted or not yet created
+   * @see #loadAtPointInTime(EppResource, DateTime)
+   */
+  public static <T extends EppResource> Supplier<T> loadAtPointInTimeAsync(
+      final T resource, final DateTime timestamp) {
+    return () -> loadAtPointInTime(resource, timestamp);
+  }
+
+  /**
+   * Returns the most recent revision of a given EppResource before or at the provided timestamp,
+   * falling back to using the resource as-is if there are no revisions.
    *
    * @see #loadAtPointInTime(EppResource, DateTime)
    */
-  private static <T extends EppResource> Result<T> loadMostRecentRevisionAtTime(
+  private static <T extends EppResource> T loadMostRecentRevisionAtTime(
+      final T resource, final DateTime timestamp) {
+    if (tm().isOfy()) {
+      return loadMostRecentRevisionAtTimeDatastore(resource, timestamp);
+    } else {
+      return loadMostRecentRevisionAtTimeSql(resource, timestamp);
+    }
+  }
+
+  /**
+   * Returns the most recent Datastore revision of a given EppResource before or at the provided
+   * timestamp using the EppResource revisions map, falling back to using the resource as-is if
+   * there are no revisions.
+   *
+   * @see #loadAtPointInTimeAsync(EppResource, DateTime)
+   */
+  private static <T extends EppResource> T loadMostRecentRevisionAtTimeDatastore(
       final T resource, final DateTime timestamp) {
     final Key<T> resourceKey = Key.create(resource);
-    final Key<CommitLogManifest> revision = findMostRecentRevisionAtTime(resource, timestamp);
+    final Key<CommitLogManifest> revision =
+        findMostRecentDatastoreRevisionAtTime(resource, timestamp);
     if (revision == null) {
       logger.atSevere().log("No revision found for %s, falling back to resource.", resourceKey);
-      return new ResultNow<>(resource);
-    }
-    final Result<CommitLogMutation> mutationResult =
-        ofy().load().key(CommitLogMutation.createKey(revision, resourceKey));
-    return () -> {
-      CommitLogMutation mutation = mutationResult.now();
-      if (mutation != null) {
-        return ofy().load().fromEntity(mutation.getEntity());
-      }
-      logger.atSevere().log(
-          "Couldn't load mutation for revision at %s for %s, falling back to resource."
-              + " Revision: %s",
-          timestamp, resourceKey, revision);
       return resource;
-    };
+    }
+    final CommitLogMutation mutation =
+        auditedOfy().load().key(CommitLogMutation.createKey(revision, resourceKey)).now();
+    if (mutation != null) {
+      return auditedOfy().load().fromEntity(mutation.getEntity());
+    }
+    logger.atSevere().log(
+        "Couldn't load mutation for revision at %s for %s, falling back to resource."
+            + " Revision: %s",
+        timestamp, resourceKey, revision);
+    return resource;
+  }
+
+  /**
+   * Returns the most recent SQL revision of a given EppResource before or at the provided timestamp
+   * using *History objects, falling back to using the resource as-is if there are no revisions.
+   *
+   * @see #loadAtPointInTimeAsync(EppResource, DateTime)
+   */
+  private static <T extends EppResource> T loadMostRecentRevisionAtTimeSql(
+      T resource, DateTime timestamp) {
+    @SuppressWarnings("unchecked")
+    T resourceAtPointInTime =
+        (T)
+            HistoryEntryDao.loadHistoryObjectsForResource(
+                    resource.createVKey(), START_OF_TIME, timestamp)
+                .stream()
+                .max(Comparator.comparing(HistoryEntry::getModificationTime))
+                .flatMap(HistoryEntry::getResourceAtPointInTime)
+                .orElse(null);
+    if (resourceAtPointInTime == null) {
+      logger.atSevere().log(
+          "Couldn't load resource at % for key %s, falling back to resource %s.",
+          timestamp, resource.createVKey(), resource);
+      return resource;
+    }
+    return resourceAtPointInTime;
   }
 
   @Nullable
-  private static <T extends EppResource> Key<CommitLogManifest>
-      findMostRecentRevisionAtTime(final T resource, final DateTime timestamp) {
+  private static <T extends EppResource>
+      Key<CommitLogManifest> findMostRecentDatastoreRevisionAtTime(
+          final T resource, final DateTime timestamp) {
     final Key<T> resourceKey = Key.create(resource);
     Entry<?, Key<CommitLogManifest>> revision = resource.getRevisions().floorEntry(timestamp);
     if (revision != null) {

core/src/main/java/google/registry/model/contact/ContactHistory.java

@@ -18,6 +18,7 @@ import static google.registry.persistence.transaction.TransactionManagerFactory.
 
 import com.googlecode.objectify.Key;
 import com.googlecode.objectify.annotation.EntitySubclass;
+import google.registry.model.EppResource;
 import google.registry.model.ImmutableObject;
 import google.registry.model.contact.ContactHistory.ContactHistoryId;
 import google.registry.model.reporting.HistoryEntry;
@@ -107,6 +108,11 @@ public class ContactHistory extends HistoryEntry implements SqlEntity {
     return (VKey<ContactHistory>) createVKey(Key.create(this));
   }
 
+  @Override
+  public Optional<? extends EppResource> getResourceAtPointInTime() {
+    return getContactBase();
+  }
+
   @PostLoad
   void postLoad() {
     // Normally Hibernate would see that the contact fields are all null and would fill contactBase

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

@@ -21,9 +21,11 @@ import static google.registry.util.CollectionUtils.nullToEmptyImmutableCopy;
 import com.google.common.collect.ImmutableSet;
 import com.googlecode.objectify.Key;
 import com.googlecode.objectify.annotation.EntitySubclass;
+import google.registry.model.EppResource;
 import google.registry.model.ImmutableObject;
 import google.registry.model.domain.DomainHistory.DomainHistoryId;
 import google.registry.model.domain.GracePeriod.GracePeriodHistory;
+import google.registry.model.domain.secdns.DelegationSignerData;
 import google.registry.model.domain.secdns.DomainDsDataHistory;
 import google.registry.model.host.HostResource;
 import google.registry.model.reporting.DomainTransactionRecord;
@@ -248,10 +250,21 @@ public class DomainHistory extends HistoryEntry implements SqlEntity {
     return (VKey<DomainHistory>) createVKey(Key.create(this));
   }
 
+  @Override
+  public Optional<? extends EppResource> getResourceAtPointInTime() {
+    return getDomainContent();
+  }
+
   @PostLoad
   void postLoad() {
     if (domainContent != null) {
       domainContent.nsHosts = nullToEmptyImmutableCopy(nsHosts);
+      domainContent.gracePeriods =
+          gracePeriodHistories.stream()
+              .map(GracePeriod::createFromHistory)
+              .collect(toImmutableSet());
+      domainContent.dsData =
+          dsDataHistories.stream().map(DelegationSignerData::create).collect(toImmutableSet());
       // Normally Hibernate would see that the domain fields are all null and would fill
       // domainContent with a null object. Unfortunately, the updateTimestamp is never null in SQL.
       if (domainContent.getDomainName() == null) {

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

@@ -115,6 +115,17 @@ public class GracePeriod extends GracePeriodBase implements DatastoreAndSqlEntit
         type, domainRepoId, expirationTime, clientId, billingEventOneTime, null, gracePeriodId);
   }
 
+  public static GracePeriod createFromHistory(GracePeriodHistory history) {
+    return createInternal(
+        history.type,
+        history.domainRepoId,
+        history.expirationTime,
+        history.clientId,
+        history.billingEventOneTime == null ? null : history.billingEventOneTime.createVKey(),
+        history.billingEventRecurring == null ? null : history.billingEventRecurring.createVKey(),
+        history.gracePeriodId);
+  }
+
   /** Creates a GracePeriod for a Recurring billing event. */
   public static GracePeriod createForRecurring(
       GracePeriodStatus type,

core/src/main/java/google/registry/model/domain/secdns/DelegationSignerData.java

@@ -114,6 +114,15 @@ public class DelegationSignerData extends DomainDsDataBase {
     return create(keyTag, algorithm, digestType, DatatypeConverter.parseHexBinary(digestAsHex));
   }
 
+  public static DelegationSignerData create(DomainDsDataHistory history) {
+    return create(
+        history.keyTag,
+        history.algorithm,
+        history.digestType,
+        history.digest,
+        history.domainRepoId);
+  }
+
   /** Class to represent the composite primary key of {@link DelegationSignerData} entity. */
   static class DomainDsDataId extends ImmutableObject implements Serializable {
 

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

@@ -18,6 +18,7 @@ import static google.registry.persistence.transaction.TransactionManagerFactory.
 
 import com.googlecode.objectify.Key;
 import com.googlecode.objectify.annotation.EntitySubclass;
+import google.registry.model.EppResource;
 import google.registry.model.ImmutableObject;
 import google.registry.model.host.HostHistory.HostHistoryId;
 import google.registry.model.reporting.HistoryEntry;
@@ -108,6 +109,11 @@ public class HostHistory extends HistoryEntry implements SqlEntity {
     return (VKey<HostHistory>) createVKey(Key.create(this));
   }
 
+  @Override
+  public Optional<? extends EppResource> getResourceAtPointInTime() {
+    return getHostBase();
+  }
+
   @PostLoad
   void postLoad() {
     // Normally Hibernate would see that the host fields are all null and would fill hostBase

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

@@ -47,6 +47,7 @@ import com.googlecode.objectify.annotation.Embed;
 import com.googlecode.objectify.annotation.Entity;
 import com.googlecode.objectify.annotation.Id;
 import com.googlecode.objectify.annotation.Mapify;
+import com.googlecode.objectify.annotation.OnLoad;
 import com.googlecode.objectify.annotation.OnSave;
 import com.googlecode.objectify.annotation.Parent;
 import google.registry.model.Buildable;
@@ -111,6 +112,26 @@ public class Registry extends ImmutableObject implements Buildable, DatastoreAnd
   @PostLoad
   void postLoad() {
     tldStr = tldStrId;
+    // TODO(sarahbot@): Remove the rest of this method after this data migration is complete
+    if (premiumListName != null) {
+      premiumList = Key.create(getCrossTldKey(), PremiumList.class, premiumListName);
+    }
+    if (reservedListNames != null) {
+      reservedLists =
+          reservedListNames.stream()
+              .map(name -> Key.create(getCrossTldKey(), ReservedList.class, name))
+              .collect(toImmutableSet());
+    }
+  }
+
+  // TODO(sarahbot@): Remove this method after this data migration is complete
+  @OnLoad
+  void onLoad() {
+    if (reservedLists != null) {
+      reservedListNames =
+          reservedLists.stream().map(key -> key.getName()).collect(toImmutableSet());
+    }
+    premiumListName = premiumList == null ? null : premiumList.getName();
   }
 
   /** The suffix that identifies roids as belonging to this specific tld, e.g. -HOW for .how. */
@@ -388,17 +409,37 @@ public class Registry extends ImmutableObject implements Buildable, DatastoreAnd
   CreateAutoTimestamp creationTime = CreateAutoTimestamp.create(null);
 
   /** The set of reserved lists that are applicable to this registry. */
-  @Column(name = "reserved_list_names")
-  Set<Key<ReservedList>> reservedLists;
+  @Transient Set<Key<ReservedList>> reservedLists;
 
-  /** Retrieves an ImmutableSet of all ReservedLists associated with this tld. */
+  /** The set of reserved list names that are applicable to this registry. */
+  @Column(name = "reserved_list_names")
+  Set<String> reservedListNames;
+
+  /**
+   * Retrieves an ImmutableSet of all ReservedLists associated with this TLD.
+   *
+   * <p>This set contains only the names of the list and not a reference to the lists. Updates to a
+   * reserved list in Cloud SQL are saved as a new ReservedList entity. When using the ReservedList
+   * for a registry, the database should be queried for the entity with this name that has the
+   * largest revision ID.
+   */
   public ImmutableSet<Key<ReservedList>> getReservedLists() {
     return nullToEmptyImmutableCopy(reservedLists);
   }
 
   /** The static {@link PremiumList} for this TLD, if there is one. */
+  @Transient Key<PremiumList> premiumList;
+
+  /**
+   * The name of the {@link PremiumList} for this TLD, if there is one.
+   *
+   * <p>This is only the name of the list and not a reference to the list. Updates to the premium
+   * list in Cloud SQL are saved as a new PremiumList entity. When using the PremiumList for a
+   * registry, the database should be queried for the entity with this name that has the largest
+   * revision ID.
+   */
   @Column(name = "premium_list_name", nullable = true)
-  Key<PremiumList> premiumList;
+  String premiumListName;
 
   /** Should RDE upload a nightly escrow deposit for this TLD? */
   @Column(nullable = false)
@@ -879,21 +920,26 @@ public class Registry extends ImmutableObject implements Buildable, DatastoreAnd
     public Builder setReservedLists(Set<ReservedList> reservedLists) {
       checkArgumentNotNull(reservedLists, "reservedLists must not be null");
       ImmutableSet.Builder<Key<ReservedList>> builder = new ImmutableSet.Builder<>();
+      ImmutableSet.Builder<String> nameBuilder = new ImmutableSet.Builder<>();
       for (ReservedList reservedList : reservedLists) {
         builder.add(Key.create(reservedList));
+        nameBuilder.add(reservedList.getName());
       }
       getInstance().reservedLists = builder.build();
+      getInstance().reservedListNames = nameBuilder.build();
       return this;
     }
 
-    public Builder setPremiumList(PremiumList premiumList) {
+    public Builder setPremiumList(@Nullable PremiumList premiumList) {
       getInstance().premiumList = (premiumList == null) ? null : Key.create(premiumList);
+      getInstance().premiumListName = (premiumList == null) ? null : premiumList.getName();
       return this;
     }
 
     @VisibleForTesting
     public Builder setPremiumListKey(@Nullable Key<PremiumList> premiumList) {
       getInstance().premiumList = premiumList;
+      getInstance().premiumListName = (premiumList == null) ? null : premiumList.getName();
       return this;
     }
 

core/src/main/java/google/registry/model/reporting/HistoryEntry.java

@@ -290,6 +290,17 @@ public class HistoryEntry extends ImmutableObject implements Buildable, Datastor
     return nullToEmptyImmutableCopy(domainTransactionRecords);
   }
 
+  /**
+   * Throws an error when attempting to retrieve the EppResource at this point in time.
+   *
+   * <p>Subclasses must override this to return the resource; it is non-abstract for legacy reasons
+   * and objects created prior to the Registry 3.0 migration.
+   */
+  public Optional<? extends EppResource> getResourceAtPointInTime() {
+    throw new UnsupportedOperationException(
+        "Raw HistoryEntry objects do not store the resource at that point in time.");
+  }
+
   /** This method exists solely to satisfy Hibernate. Use the {@link Builder} instead. */
   @SuppressWarnings("UnusedMethod")
   private void setPeriod(Period period) {

core/src/main/java/google/registry/module/backend/BackendRequestComponent.java

@@ -20,6 +20,7 @@ import google.registry.backup.BackupModule;
 import google.registry.backup.CommitLogCheckpointAction;
 import google.registry.backup.DeleteOldCommitLogsAction;
 import google.registry.backup.ExportCommitLogDiffAction;
+import google.registry.backup.ReplayCommitLogsToSqlAction;
 import google.registry.batch.BatchModule;
 import google.registry.batch.DeleteContactsAndHostsAction;
 import google.registry.batch.DeleteExpiredDomainsAction;
@@ -186,6 +187,8 @@ interface BackendRequestComponent {
 
   RelockDomainAction relockDomainAction();
 
+  ReplayCommitLogsToSqlAction replayCommitLogsToSqlAction();
+
   ResaveAllEppResourcesAction resaveAllEppResourcesAction();
 
   ResaveEntityAction resaveEntityAction();

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

@@ -17,6 +17,7 @@ package google.registry.rde;
 import static com.google.common.base.Strings.nullToEmpty;
 import static com.google.common.collect.ImmutableSet.toImmutableSet;
 import static google.registry.model.EppResourceUtils.loadAtPointInTime;
+import static google.registry.model.EppResourceUtils.loadAtPointInTimeAsync;
 import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
 
 import com.google.appengine.tools.mapreduce.Mapper;
@@ -26,7 +27,6 @@ import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.ImmutableSetMultimap;
 import com.google.common.collect.Maps;
 import com.google.common.collect.Sets;
-import com.googlecode.objectify.Result;
 import google.registry.model.EppResource;
 import google.registry.model.contact.ContactResource;
 import google.registry.model.domain.DomainBase;
@@ -36,7 +36,9 @@ import google.registry.model.registrar.Registrar;
 import google.registry.xml.ValidationMode;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Optional;
+import java.util.function.Supplier;
 import org.joda.time.DateTime;
 
 /** Mapper for {@link RdeStagingAction}. */
@@ -123,8 +125,8 @@ public final class RdeStagingMapper extends Mapper<EppResource, PendingDeposit,
             .collect(toImmutableSet());
 
     // Launch asynchronous fetches of point-in-time representations of resource.
-    ImmutableMap<DateTime, Result<EppResource>> resourceAtTimes =
-        ImmutableMap.copyOf(Maps.asMap(dates, input -> loadAtPointInTime(resource, input)));
+    ImmutableMap<DateTime, Supplier<EppResource>> resourceAtTimes =
+        ImmutableMap.copyOf(Maps.asMap(dates, input -> loadAtPointInTimeAsync(resource, input)));
 
     // Convert resource to an XML fragment for each watermark/mode pair lazily and cache the result.
     Fragmenter fragmenter = new Fragmenter(resourceAtTimes);
@@ -159,13 +161,13 @@ public final class RdeStagingMapper extends Mapper<EppResource, PendingDeposit,
   /** Loading cache that turns a resource into XML for the various points in time and modes. */
   private class Fragmenter {
     private final Map<WatermarkModePair, Optional<DepositFragment>> cache = new HashMap<>();
-    private final ImmutableMap<DateTime, Result<EppResource>> resourceAtTimes;
+    private final ImmutableMap<DateTime, Supplier<EppResource>> resourceAtTimes;
 
     long cacheHits = 0;
     long resourcesNotFound = 0;
     long resourcesFound = 0;
 
-    Fragmenter(ImmutableMap<DateTime, Result<EppResource>> resourceAtTimes) {
+    Fragmenter(ImmutableMap<DateTime, Supplier<EppResource>> resourceAtTimes) {
       this.resourceAtTimes = resourceAtTimes;
     }
 
@@ -175,7 +177,7 @@ public final class RdeStagingMapper extends Mapper<EppResource, PendingDeposit,
         cacheHits++;
         return result;
       }
-      EppResource resource = resourceAtTimes.get(watermark).now();
+      EppResource resource = resourceAtTimes.get(watermark).get();
       if (resource == null) {
         result = Optional.empty();
         cache.put(WatermarkModePair.create(watermark, RdeMode.FULL), result);
@@ -202,8 +204,9 @@ public final class RdeStagingMapper extends Mapper<EppResource, PendingDeposit,
                         host,
                         // Note that loadAtPointInTime() does cloneProjectedAtTime(watermark) for
                         // us.
-                        loadAtPointInTime(tm().loadByKey(host.getSuperordinateDomain()), watermark)
-                            .now())
+                        Objects.requireNonNull(
+                            loadAtPointInTime(
+                                tm().loadByKey(host.getSuperordinateDomain()), watermark)))
                     : marshaller.marshalExternalHost(host));
         cache.put(WatermarkModePair.create(watermark, RdeMode.FULL), result);
         cache.put(WatermarkModePair.create(watermark, RdeMode.THIN), result);

core/src/main/java/google/registry/reporting/ReportingModule.java

@@ -14,6 +14,7 @@
 
 package google.registry.reporting;
 
+import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
 import static google.registry.request.RequestParameters.extractOptionalParameter;
 import static google.registry.request.RequestParameters.extractRequiredParameter;
 
@@ -55,6 +56,9 @@ public class ReportingModule {
   /** The request parameter specifying the jobId for a running Dataflow pipeline. */
   public static final String PARAM_JOB_ID = "jobId";
 
+  /** The request parameter for specifying which database reporting actions should read from. */
+  public static final String DATABASE = "database";
+
   /** Provides the Cloud Dataflow jobId for a pipeline. */
   @Provides
   @Parameter(PARAM_JOB_ID)
@@ -62,6 +66,14 @@ public class ReportingModule {
     return extractRequiredParameter(req, PARAM_JOB_ID);
   }
 
+  /** Provides the database for the pipeline to read from. */
+  @Provides
+  @Parameter(DATABASE)
+  static String provideDatabase(HttpServletRequest req) {
+    Optional<String> optionalDatabase = extractOptionalParameter(req, DATABASE);
+    return optionalDatabase.orElse(tm().isOfy() ? "DATASTORE" : "CLOUD_SQL");
+  }
+
   /** Extracts an optional YearMonth in yyyy-MM format from the request. */
   @Provides
   @Parameter(PARAM_YEAR_MONTH)

core/src/main/java/google/registry/reporting/spec11/GenerateSpec11ReportAction.java

@@ -15,6 +15,8 @@
 package google.registry.reporting.spec11;
 
 import static google.registry.beam.BeamUtils.createJobName;
+import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
+import static google.registry.reporting.ReportingModule.DATABASE;
 import static google.registry.reporting.ReportingModule.PARAM_DATE;
 import static google.registry.reporting.ReportingUtils.enqueueBeamReportingTask;
 import static google.registry.request.Action.Method.POST;
@@ -69,6 +71,7 @@ public class GenerateSpec11ReportAction implements Runnable {
   private final Clock clock;
   private final Response response;
   private final Dataflow dataflow;
+  private final String database;
 
   @Inject
   GenerateSpec11ReportAction(
@@ -78,15 +81,20 @@ public class GenerateSpec11ReportAction implements Runnable {
       @Config("reportingBucketUrl") String reportingBucketUrl,
       @Key("safeBrowsingAPIKey") String apiKey,
       @Parameter(PARAM_DATE) LocalDate date,
+      @Parameter(DATABASE) String database,
       Clock clock,
       Response response,
       Dataflow dataflow) {
     this.projectId = projectId;
     this.jobRegion = jobRegion;
     this.stagingBucketUrl = stagingBucketUrl;
+    if (tm().isOfy() && database.equals("CLOUD_SQL")) {
+      reportingBucketUrl = reportingBucketUrl.concat("-sql");
+    }
     this.reportingBucketUrl = reportingBucketUrl;
     this.apiKey = apiKey;
     this.date = date;
+    this.database = database;
     this.clock = clock;
     this.response = response;
     this.dataflow = dataflow;
@@ -105,6 +113,8 @@ public class GenerateSpec11ReportAction implements Runnable {
                   ImmutableMap.of(
                       "safeBrowsingApiKey",
                       apiKey,
+                      "database",
+                      database,
                       ReportingModule.PARAM_DATE,
                       date.toString(),
                       "reportingBucketUrl",

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

@@ -80,8 +80,7 @@ final class CreateDomainCommand extends CreateOrUpdateDomainCommand
       }
 
       setSoyTemplate(DomainCreateSoyInfo.getInstance(), DomainCreateSoyInfo.DOMAINCREATE);
-      addSoyRecord(
-          clientId,
+      SoyMapData soyMapData =
           new SoyMapData(
               "domain", domain,
               "period", period,
@@ -92,7 +91,12 @@ final class CreateDomainCommand extends CreateOrUpdateDomainCommand
               "password", password,
               "currency", currency,
               "price", cost,
-              "dsRecords", DsRecord.convertToSoy(dsRecords)));
+              "dsRecords", DsRecord.convertToSoy(dsRecords),
+              "reason", reason);
+      if (requestedByRegistrar != null) {
+        soyMapData.put("requestedByRegistrar", requestedByRegistrar.toString());
+      }
+      addSoyRecord(clientId, soyMapData);
     }
   }
 }

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

@@ -78,6 +78,17 @@ abstract class CreateOrUpdateDomainCommand extends MutatingEppToolCommand {
       converter = DsRecord.Converter.class)
   List<DsRecord> dsRecords = new ArrayList<>();
 
+  @Parameter(
+      names = {"--reason"},
+      description = "Reason for the change.")
+  String reason;
+
+  @Parameter(
+      names = {"--registrar_request"},
+      description = "Whether the change was requested by a registrar.",
+      arity = 1)
+  Boolean requestedByRegistrar;
+
   Set<String> domains;
 
   @Override

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

@@ -20,6 +20,7 @@ import google.registry.tools.javascrap.BackfillSpec11ThreatMatchesCommand;
 import google.registry.tools.javascrap.DeleteContactByRoidCommand;
 import google.registry.tools.javascrap.PopulateNullRegistrarFieldsCommand;
 import google.registry.tools.javascrap.RemoveIpAddressCommand;
+import google.registry.tools.javascrap.ResaveAllTldsCommand;
 
 /** Container class to create and run remote commands against a Datastore instance. */
 public final class RegistryTool {
@@ -106,6 +107,7 @@ public final class RegistryTool {
           .put("remove_ip_address", RemoveIpAddressCommand.class)
           .put("remove_registry_one_key", RemoveRegistryOneKeyCommand.class)
           .put("renew_domain", RenewDomainCommand.class)
+          .put("resave_all_tlds", ResaveAllTldsCommand.class)
           .put("resave_entities", ResaveEntitiesCommand.class)
           .put("resave_environment_entities", ResaveEnvironmentEntitiesCommand.class)
           .put("resave_epp_resource", ResaveEppResourceCommand.class)

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

@@ -315,10 +315,14 @@ final class UpdateDomainCommand extends CreateOrUpdateDomainCommand {
               "secdns", secDns,
               "addDsRecords", DsRecord.convertToSoy(addDsRecords),
               "removeDsRecords", DsRecord.convertToSoy(removeDsRecords),
-              "removeAllDsRecords", clearDsRecords);
+              "removeAllDsRecords", clearDsRecords,
+              "reason", reason);
       if (autorenews != null) {
         soyMapData.put("autorenews", autorenews.toString());
       }
+      if (requestedByRegistrar != null) {
+        soyMapData.put("requestedByRegistrar", requestedByRegistrar.toString());
+      }
       addSoyRecord(clientId, soyMapData);
     }
 

core/src/main/java/google/registry/tools/javascrap/ResaveAllTldsCommand.java

@@ -0,0 +1,30 @@
+// Copyright 2021 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.tools.javascrap;
+
+import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
+
+import com.beust.jcommander.Parameters;
+import google.registry.model.registry.Registry;
+import google.registry.tools.CommandWithRemoteApi;
+
+/** Scrap command to resave all Registry entities. */
+@Parameters(commandDescription = "Resave all TLDs")
+public class ResaveAllTldsCommand implements CommandWithRemoteApi {
+  @Override
+  public void run() throws Exception {
+    tm().transact(() -> tm().putAll(tm().loadAllOf(Registry.class)));
+  }
+}

core/src/main/java/google/registry/tools/server/GenerateZoneFilesAction.java

@@ -198,7 +198,7 @@ public class GenerateZoneFilesAction implements Runnable, JsonActionRunner.JsonA
     private void mapDomain(DomainBase domain) {
       // Domains never change their tld, so we can check if it's from the wrong tld right away.
       if (tlds.contains(domain.getTld())) {
-        domain = loadAtPointInTime(domain, exportTime).now();
+        domain = loadAtPointInTime(domain, exportTime);
         // A null means the domain was deleted (or not created) at this time.
         if (domain != null && domain.shouldPublishToDns()) {
           String stanza = domainStanza(domain, exportTime, dnsDefaultNsTtl, dnsDefaultDsTtl);
@@ -215,7 +215,7 @@ public class GenerateZoneFilesAction implements Runnable, JsonActionRunner.JsonA
       ImmutableSet<String> subordinateHosts = domain.getSubordinateHosts();
       if (!subordinateHosts.isEmpty()) {
         for (HostResource unprojectedHost : tm().loadByKeys(domain.getNameservers()).values()) {
-          HostResource host = loadAtPointInTime(unprojectedHost, exportTime).now();
+          HostResource host = loadAtPointInTime(unprojectedHost, exportTime);
           // A null means the host was deleted (or not created) at this time.
           if ((host != null) && subordinateHosts.contains(host.getHostName())) {
             String stanza = hostStanza(host, dnsDefaultATtl, domain.getTld());
@@ -290,7 +290,7 @@ public class GenerateZoneFilesAction implements Runnable, JsonActionRunner.JsonA
               domainLabel,
               dnsDefaultNsTtl.getStandardSeconds(),
               // Load the nameservers at the export time in case they've been renamed or deleted.
-              loadAtPointInTime(nameserver, exportTime).now().getHostName()));
+              loadAtPointInTime(nameserver, exportTime).getHostName()));
     }
     for (DelegationSignerData dsData : domain.getDsData()) {
       result.append(

core/src/main/java/google/registry/tools/server/KillAllCommitLogsAction.java

@@ -49,6 +49,8 @@ import javax.inject.Inject;
     path = "/_dr/task/killAllCommitLogs",
     method = POST,
     auth = Auth.AUTH_INTERNAL_OR_ADMIN)
+// No longer needed in SQL. Subject to future removal.
+@Deprecated
 public class KillAllCommitLogsAction implements Runnable {
 
   @Inject MapreduceRunner mrRunner;

core/src/main/java/google/registry/tools/server/RefreshDnsForAllDomainsAction.java

@@ -18,6 +18,8 @@ import static com.google.common.base.Preconditions.checkArgument;
 import static google.registry.mapreduce.inputs.EppResourceInputs.createEntityInput;
 import static google.registry.model.EppResourceUtils.isActive;
 import static google.registry.model.registry.Registries.assertTldsExist;
+import static google.registry.persistence.transaction.TransactionManagerFactory.jpaTm;
+import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
 import static google.registry.request.RequestParameters.PARAM_TLDS;
 
 import com.google.appengine.tools.mapreduce.Mapper;
@@ -32,11 +34,12 @@ import google.registry.request.Action;
 import google.registry.request.Parameter;
 import google.registry.request.Response;
 import google.registry.request.auth.Auth;
+import google.registry.util.Clock;
 import google.registry.util.NonFinalForTesting;
 import java.util.Random;
 import javax.inject.Inject;
+import org.apache.http.HttpStatus;
 import org.joda.time.DateTime;
-import org.joda.time.DateTimeZone;
 import org.joda.time.Duration;
 
 /**
@@ -74,6 +77,8 @@ public class RefreshDnsForAllDomainsAction implements Runnable {
   @Parameter("smearMinutes")
   int smearMinutes;
 
+  @Inject DnsQueue dnsQueue;
+  @Inject Clock clock;
   @Inject Random random;
 
   @Inject
@@ -83,14 +88,41 @@ public class RefreshDnsForAllDomainsAction implements Runnable {
   public void run() {
     assertTldsExist(tlds);
     checkArgument(smearMinutes > 0, "Must specify a positive number of smear minutes");
-    mrRunner
-        .setJobName("Refresh DNS for all domains")
-        .setModuleName("tools")
-        .setDefaultMapShards(10)
-        .runMapOnly(
-            new RefreshDnsForAllDomainsActionMapper(tlds, smearMinutes, random),
-            ImmutableList.of(createEntityInput(DomainBase.class)))
-        .sendLinkToMapreduceConsole(response);
+    if (tm().isOfy()) {
+      mrRunner
+          .setJobName("Refresh DNS for all domains")
+          .setModuleName("tools")
+          .setDefaultMapShards(10)
+          .runMapOnly(
+              new RefreshDnsForAllDomainsActionMapper(tlds, smearMinutes, random, clock.nowUtc()),
+              ImmutableList.of(createEntityInput(DomainBase.class)))
+          .sendLinkToMapreduceConsole(response);
+    } else {
+      tm().transact(
+              () ->
+                  jpaTm()
+                      .query(
+                          "SELECT fullyQualifiedDomainName FROM Domain "
+                              + "WHERE tld IN (:tlds) "
+                              + "AND deletionTime > :now",
+                          String.class)
+                      .setParameter("tlds", tlds)
+                      .setParameter("now", clock.nowUtc())
+                      .getResultStream()
+                      .forEach(
+                          domainName -> {
+                            try {
+                              // Smear the task execution time over the next N minutes.
+                              dnsQueue.addDomainRefreshTask(
+                                  domainName,
+                                  Duration.standardMinutes(random.nextInt(smearMinutes)));
+                            } catch (Throwable t) {
+                              logger.atSevere().withCause(t).log(
+                                  "Error while enqueuing DNS refresh for domain %s", domainName);
+                              response.setStatus(HttpStatus.SC_INTERNAL_SERVER_ERROR);
+                            }
+                          }));
+    }
   }
 
   /** Mapper to refresh DNS for all active domain resources. */
@@ -104,19 +136,21 @@ public class RefreshDnsForAllDomainsAction implements Runnable {
     private final ImmutableSet<String> tlds;
     private final int smearMinutes;
     private final Random random;
+    private final DateTime now;
 
     RefreshDnsForAllDomainsActionMapper(
-        ImmutableSet<String> tlds, int smearMinutes, Random random) {
+        ImmutableSet<String> tlds, int smearMinutes, Random random, DateTime now) {
       this.tlds = tlds;
       this.smearMinutes = smearMinutes;
       this.random = random;
+      this.now = now;
     }
 
     @Override
     public void map(final DomainBase domain) {
       String domainName = domain.getDomainName();
       if (tlds.contains(domain.getTld())) {
-        if (isActive(domain, DateTime.now(DateTimeZone.UTC))) {
+        if (isActive(domain, now)) {
           try {
             // Smear the task execution time over the next N minutes.
             dnsQueue.addDomainRefreshTask(
@@ -124,7 +158,7 @@ public class RefreshDnsForAllDomainsAction implements Runnable {
             getContext().incrementCounter("active domains refreshed");
           } catch (Throwable t) {
             logger.atSevere().withCause(t).log(
-                "Error while refreshing DNS for domain %s", domainName);
+                "Error while enqueuing DNS refresh for domain %s", domainName);
             getContext().incrementCounter("active domains errored");
           }
         } else {

core/src/main/java/google/registry/tools/server/ResaveAllHistoryEntriesAction.java

@@ -42,6 +42,8 @@ import javax.inject.Inject;
     service = Action.Service.TOOLS,
     path = "/_dr/task/resaveAllHistoryEntries",
     auth = Auth.AUTH_INTERNAL_OR_ADMIN)
+// No longer needed in SQL. Subject to future removal.
+@Deprecated
 public class ResaveAllHistoryEntriesAction implements Runnable {
 
   @Inject MapreduceRunner mrRunner;

core/src/main/resources/google/registry/beam/spec11_pipeline_metadata.json

@@ -61,6 +61,14 @@
       "regexes": [
         "^gs:\\/\\/[^\\n\\r]+$"
       ]
+    },
+    {
+      "name": "database",
+      "label": "Database to read from.",
+      "helpText": "DATASTORE or CLOUD_SQL.",
+      "regexes": [
+        "^DATASTORE|CLOUD_SQL$"
+      ]
     }
   ]
 }

core/src/main/resources/google/registry/tools/soy/DomainCreate.soy

@@ -27,6 +27,8 @@
   {@param? currency: string}
   {@param? price: string}
   {@param dsRecords: list<[keyTag:int, alg:int, digestType:int, digest:string]>}
+  {@param? reason: string}
+  {@param? requestedByRegistrar: string}
 
   <?xml version="1.0" encoding="UTF-8" standalone="yes"?>
   <epp xmlns="urn:ietf:params:xml:ns:epp-1.0">
@@ -54,7 +56,7 @@
           </domain:authInfo>
         </domain:create>
       </create>
-      {if length($dsRecords) > 0 or $price != null}
+      {if length($dsRecords) > 0 or $price != null or $reason or $requestedByRegistrar}
         <extension>
           {if $price != null}
             <fee:create xmlns:fee="urn:ietf:params:xml:ns:fee-0.12">
@@ -74,6 +76,16 @@
               {/for}
             </secDNS:create>
           {/if}
+          {if $reason or $requestedByRegistrar}
+            <metadata:metadata xmlns:metadata="urn:google:params:xml:ns:metadata-1.0">
+            {if $reason}
+              <metadata:reason>{$reason}</metadata:reason>
+            {/if}
+            {if $requestedByRegistrar}
+              <metadata:requestedByRegistrar>{$requestedByRegistrar}</metadata:requestedByRegistrar>
+            {/if}
+            </metadata:metadata>
+          {/if}
         </extension>
       {/if}
       <clTRID>RegistryTool</clTRID>

core/src/main/resources/google/registry/tools/soy/DomainUpdate.soy

@@ -36,6 +36,8 @@
   {@param removeDsRecords: list<[keyTag:int, alg:int, digestType:int, digest:string]>}
   {@param removeAllDsRecords: bool}
   {@param? autorenews: string}
+  {@param? reason: string}
+  {@param? requestedByRegistrar: string}
 
   <?xml version="1.0" encoding="UTF-8" standalone="yes"?>
   <epp xmlns="urn:ietf:params:xml:ns:epp-1.0">
@@ -97,7 +99,7 @@
           {/if}
         </domain:update>
       </update>
-      {if $secdns or $autorenews}
+      {if $secdns or $autorenews or $reason or $requestedByRegistrar}
         <extension>
           {if $secdns}
             <secDNS:update xmlns:secDNS="urn:ietf:params:xml:ns:secDNS-1.1">
@@ -137,6 +139,16 @@
               <superuser:autorenews>{$autorenews}</superuser:autorenews>
             </superuser:domainUpdate>
           {/if}
+          {if $reason or $requestedByRegistrar}
+            <metadata:metadata xmlns:metadata="urn:google:params:xml:ns:metadata-1.0">
+            {if $reason}
+              <metadata:reason>{$reason}</metadata:reason>
+            {/if}
+            {if $requestedByRegistrar}
+              <metadata:requestedByRegistrar>{$requestedByRegistrar}</metadata:requestedByRegistrar>
+            {/if}
+            </metadata:metadata>
+          {/if}
         </extension>
       {/if}
       <clTRID>RegistryTool</clTRID>

core/src/test/java/google/registry/beam/spec11/SafeBrowsingTransformsTest.java

@@ -164,7 +164,7 @@ class SafeBrowsingTransformsTest {
    * A serializable {@link Answer} that returns a mock HTTP response based on the HTTP request's
    * content.
    */
-  private static class HttpResponder implements Answer<CloseableHttpResponse>, Serializable {
+  static class HttpResponder implements Answer<CloseableHttpResponse>, Serializable {
     @Override
     public CloseableHttpResponse answer(InvocationOnMock invocation) throws Throwable {
       return getMockResponse(

core/src/test/java/google/registry/beam/spec11/Spec11PipelineTest.java

@@ -18,33 +18,60 @@ import static com.google.common.collect.ImmutableList.toImmutableList;
 import static com.google.common.truth.Truth.assertThat;
 import static google.registry.model.ImmutableObjectSubject.immutableObjectCorrespondence;
 import static google.registry.persistence.transaction.TransactionManagerFactory.jpaTm;
+import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
+import static google.registry.testing.AppEngineExtension.makeRegistrar1;
+import static google.registry.testing.DatabaseHelper.createTld;
+import static google.registry.testing.DatabaseHelper.persistActiveContact;
+import static google.registry.testing.DatabaseHelper.persistNewRegistrar;
+import static google.registry.testing.DatabaseHelper.persistResource;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.mockito.Mockito.withSettings;
 
+import com.google.common.base.Suppliers;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.Streams;
 import com.google.common.truth.Correspondence;
 import com.google.common.truth.Correspondence.BinaryPredicate;
 import google.registry.beam.TestPipelineExtension;
+import google.registry.beam.spec11.SafeBrowsingTransforms.EvaluateSafeBrowsingFn;
+import google.registry.beam.spec11.SafeBrowsingTransformsTest.HttpResponder;
+import google.registry.model.contact.ContactResource;
+import google.registry.model.domain.DomainAuthInfo;
+import google.registry.model.domain.DomainBase;
+import google.registry.model.eppcommon.AuthInfo.PasswordAuth;
+import google.registry.model.registrar.Registrar;
 import google.registry.model.reporting.Spec11ThreatMatch;
 import google.registry.model.reporting.Spec11ThreatMatch.ThreatType;
 import google.registry.model.reporting.Spec11ThreatMatchDao;
 import google.registry.persistence.transaction.JpaTestRules;
 import google.registry.persistence.transaction.JpaTestRules.JpaIntegrationTestExtension;
+import google.registry.persistence.transaction.TransactionManager;
+import google.registry.persistence.transaction.TransactionManagerFactory;
 import google.registry.testing.DatastoreEntityExtension;
 import google.registry.testing.FakeClock;
+import google.registry.testing.FakeSleeper;
 import google.registry.util.ResourceUtils;
+import google.registry.util.Retrier;
 import java.io.File;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import org.apache.beam.sdk.coders.KvCoder;
 import org.apache.beam.sdk.coders.SerializableCoder;
 import org.apache.beam.sdk.options.PipelineOptionsFactory;
+import org.apache.beam.sdk.testing.PAssert;
 import org.apache.beam.sdk.transforms.Create;
 import org.apache.beam.sdk.values.KV;
 import org.apache.beam.sdk.values.PCollection;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.impl.client.CloseableHttpClient;
 import org.checkerframework.checker.nullness.qual.Nullable;
+import org.joda.time.DateTime;
 import org.joda.time.LocalDate;
 import org.json.JSONObject;
+import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Order;
 import org.junit.jupiter.api.Test;
@@ -60,9 +87,14 @@ import org.junit.jupiter.api.io.TempDir;
  */
 class Spec11PipelineTest {
 
+  private static final DateTime START_TIME = DateTime.parse("2020-01-27T00:00:00.0Z");
+  private final FakeClock fakeClock = new FakeClock(START_TIME);
+
   private static final String DATE = "2020-01-27";
   private static final String SAFE_BROWSING_API_KEY = "api-key";
   private static final String REPORTING_BUCKET_URL = "reporting_bucket";
+  private final CloseableHttpClient mockHttpClient =
+      mock(CloseableHttpClient.class, withSettings().serializable());
 
   private static final ImmutableList<Subdomain> SUBDOMAINS =
       ImmutableList.of(
@@ -103,12 +135,18 @@ class Spec11PipelineTest {
   private File reportingBucketUrl;
   private PCollection<KV<Subdomain, ThreatMatch>> threatMatches;
 
+  ImmutableSet<Spec11ThreatMatch> sqlThreatMatches;
+  TransactionManager tm;
+
   @BeforeEach
   void beforeEach() throws Exception {
+    tm = tm();
+    TransactionManagerFactory.setTm(jpaTm());
     reportingBucketUrl = Files.createDirectory(tmpDir.resolve(REPORTING_BUCKET_URL)).toFile();
     options.setDate(DATE);
     options.setSafeBrowsingApiKey(SAFE_BROWSING_API_KEY);
     options.setReportingBucketUrl(reportingBucketUrl.getAbsolutePath());
+    options.setDatabase("DATASTORE");
     threatMatches =
         pipeline.apply(
             Create.of(
@@ -118,11 +156,8 @@ class Spec11PipelineTest {
                     KvCoder.of(
                         SerializableCoder.of(Subdomain.class),
                         SerializableCoder.of(ThreatMatch.class))));
-  }
 
-  @Test
-  void testSuccess_saveToSql() {
-    ImmutableSet<Spec11ThreatMatch> sqlThreatMatches =
+    sqlThreatMatches =
         ImmutableSet.of(
             new Spec11ThreatMatch.Builder()
                 .setDomainName("111.com")
@@ -159,25 +194,98 @@ class Spec11PipelineTest {
                 .setCheckDate(new LocalDate(2020, 1, 27))
                 .setThreatTypes(ImmutableSet.of(ThreatType.UNWANTED_SOFTWARE))
                 .build());
+  }
+
+  @AfterEach
+  void afterEach() {
+    TransactionManagerFactory.setTm(tm);
+  }
+
+  @Test
+  void testSuccess_fullSqlPipeline() throws Exception {
+    setupCloudSql();
+    options.setDatabase("CLOUD_SQL");
+    EvaluateSafeBrowsingFn safeBrowsingFn =
+        new EvaluateSafeBrowsingFn(
+            SAFE_BROWSING_API_KEY,
+            new Retrier(new FakeSleeper(new FakeClock()), 1),
+            Suppliers.ofInstance(mockHttpClient));
+    when(mockHttpClient.execute(any(HttpPost.class))).thenAnswer(new HttpResponder());
+    Spec11Pipeline spec11Pipeline = new Spec11Pipeline(options, safeBrowsingFn);
+    spec11Pipeline.setupPipeline(pipeline);
+    pipeline.run(options).waitUntilFinish();
+    verifySaveToGcs();
+    verifySaveToCloudSql();
+  }
+
+  @Test
+  void testSuccess_saveToSql() {
     Spec11Pipeline.saveToSql(threatMatches, options);
     pipeline.run().waitUntilFinish();
-    assertThat(
-            jpaTm()
-                .transact(
-                    () ->
-                        Spec11ThreatMatchDao.loadEntriesByDate(
-                            jpaTm(), new LocalDate(2020, 1, 27))))
-        .comparingElementsUsing(immutableObjectCorrespondence("id"))
-        .containsExactlyElementsIn(sqlThreatMatches);
+    verifySaveToCloudSql();
   }
 
   @Test
   void testSuccess_saveToGcs() throws Exception {
+    Spec11Pipeline.saveToGcs(threatMatches, options);
+    pipeline.run().waitUntilFinish();
+    verifySaveToGcs();
+  }
+
+  @Test
+  void testSuccess_readFromCloudSql() throws Exception {
+    setupCloudSql();
+    PCollection<Subdomain> subdomains = Spec11Pipeline.readFromCloudSql(pipeline);
+    PAssert.that(subdomains).containsInAnyOrder(SUBDOMAINS);
+    pipeline.run().waitUntilFinish();
+  }
+
+  private void setupCloudSql() {
+    persistNewRegistrar("TheRegistrar");
+    persistNewRegistrar("NewRegistrar");
+    Registrar registrar1 =
+        persistResource(
+            makeRegistrar1()
+                .asBuilder()
+                .setClientId("hello-registrar")
+                .setEmailAddress("email@hello.net")
+                .build());
+    Registrar registrar2 =
+        persistResource(
+            makeRegistrar1()
+                .asBuilder()
+                .setClientId("kitty-registrar")
+                .setEmailAddress("contact@kit.ty")
+                .build());
+    Registrar registrar3 =
+        persistResource(
+            makeRegistrar1()
+                .asBuilder()
+                .setClientId("cool-registrar")
+                .setEmailAddress("cool@aid.net")
+                .build());
+
+    createTld("com");
+    createTld("net");
+    createTld("bank");
+    createTld("dev");
+
+    ContactResource contact1 = persistActiveContact(registrar1.getClientId());
+    ContactResource contact2 = persistActiveContact(registrar2.getClientId());
+    ContactResource contact3 = persistActiveContact(registrar3.getClientId());
+
+    persistResource(createDomain("111.com", "123456789-COM", registrar1, contact1));
+    persistResource(createDomain("party-night.net", "2244AABBC-NET", registrar2, contact2));
+    persistResource(createDomain("bitcoin.bank", "1C3D5E7F9-BANK", registrar1, contact1));
+    persistResource(createDomain("no-email.com", "2A4BA9BBC-COM", registrar2, contact2));
+    persistResource(
+        createDomain("anti-anti-anti-virus.dev", "555666888-DEV", registrar3, contact3));
+  }
+
+  private void verifySaveToGcs() throws Exception {
     ImmutableList<String> expectedFileContents =
         ImmutableList.copyOf(
             ResourceUtils.readResourceUtf8(this.getClass(), "test_output.txt").split("\n"));
-    Spec11Pipeline.saveToGcs(threatMatches, options);
-    pipeline.run().waitUntilFinish();
     ImmutableList<String> resultFileContents = resultFileContents();
     assertThat(resultFileContents.size()).isEqualTo(expectedFileContents.size());
     assertThat(resultFileContents.get(0)).isEqualTo(expectedFileContents.get(0));
@@ -188,6 +296,34 @@ class Spec11PipelineTest {
         .containsExactlyElementsIn(expectedFileContents.subList(1, expectedFileContents.size()));
   }
 
+  private void verifySaveToCloudSql() {
+    jpaTm()
+        .transact(
+            () -> {
+              ImmutableList<Spec11ThreatMatch> sqlThreatMatches =
+                  Spec11ThreatMatchDao.loadEntriesByDate(jpaTm(), new LocalDate(2020, 1, 27));
+              assertThat(sqlThreatMatches)
+                  .comparingElementsUsing(immutableObjectCorrespondence("id"))
+                  .containsExactlyElementsIn(sqlThreatMatches);
+            });
+  }
+
+  private DomainBase createDomain(
+      String domainName, String repoId, Registrar registrar, ContactResource contact) {
+    return new DomainBase.Builder()
+        .setDomainName(domainName)
+        .setRepoId(repoId)
+        .setCreationClientId(registrar.getClientId())
+        .setLastEppUpdateTime(fakeClock.nowUtc())
+        .setLastEppUpdateClientId(registrar.getClientId())
+        .setLastTransferTime(fakeClock.nowUtc())
+        .setRegistrant(contact.createVKey())
+        .setPersistedCurrentSponsorClientId(registrar.getClientId())
+        .setRegistrationExpirationTime(fakeClock.nowUtc().plusYears(1))
+        .setAuthInfo(DomainAuthInfo.create(PasswordAuth.create("password")))
+        .build();
+  }
+
   /** Returns the text contents of a file under the beamBucket/results directory. */
   private ImmutableList<String> resultFileContents() throws Exception {
     File resultFile =

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

@@ -16,8 +16,7 @@ package google.registry.export;
 
 import static com.google.appengine.tools.cloudstorage.GcsServiceFactory.createGcsService;
 import static com.google.common.truth.Truth.assertThat;
-import static google.registry.export.ExportDomainListsAction.ExportDomainListsReducer.EXPORT_MIME_TYPE;
-import static google.registry.export.ExportDomainListsAction.ExportDomainListsReducer.REGISTERED_DOMAINS_FILENAME;
+import static google.registry.export.ExportDomainListsAction.REGISTERED_DOMAINS_FILENAME;
 import static google.registry.testing.DatabaseHelper.createTld;
 import static google.registry.testing.DatabaseHelper.persistActiveDomain;
 import static google.registry.testing.DatabaseHelper.persistDeletedDomain;
@@ -34,25 +33,40 @@ import com.google.appengine.tools.cloudstorage.GcsFilename;
 import com.google.appengine.tools.cloudstorage.GcsService;
 import com.google.appengine.tools.cloudstorage.ListOptions;
 import com.google.appengine.tools.cloudstorage.ListResult;
+import com.google.common.net.MediaType;
 import google.registry.export.ExportDomainListsAction.ExportDomainListsReducer;
+import google.registry.model.ofy.Ofy;
 import google.registry.model.registry.Registry;
 import google.registry.model.registry.Registry.TldType;
 import google.registry.storage.drive.DriveConnection;
+import google.registry.testing.DualDatabaseTest;
+import google.registry.testing.FakeClock;
 import google.registry.testing.FakeResponse;
+import google.registry.testing.InjectExtension;
+import google.registry.testing.TestOfyAndSql;
+import google.registry.testing.TestOfyOnly;
 import google.registry.testing.mapreduce.MapreduceTestCase;
 import java.io.FileNotFoundException;
 import org.joda.time.DateTime;
 import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Order;
+import org.junit.jupiter.api.extension.RegisterExtension;
 import org.mockito.ArgumentCaptor;
 
 /** Unit tests for {@link ExportDomainListsAction}. */
+@DualDatabaseTest
 class ExportDomainListsActionTest extends MapreduceTestCase<ExportDomainListsAction> {
 
   private GcsService gcsService;
   private DriveConnection driveConnection = mock(DriveConnection.class);
   private ArgumentCaptor<byte[]> bytesExportedToDrive = ArgumentCaptor.forClass(byte[].class);
   private final FakeResponse response = new FakeResponse();
+  private final FakeClock clock = new FakeClock(DateTime.parse("2020-02-02T02:02:02Z"));
+
+  @Order(Order.DEFAULT - 1)
+  @RegisterExtension
+  public final InjectExtension inject =
+      new InjectExtension().withStaticFieldOverride(Ofy.class, "clock", clock);
 
   @BeforeEach
   void beforeEach() {
@@ -68,10 +82,12 @@ class ExportDomainListsActionTest extends MapreduceTestCase<ExportDomainListsAct
     action.response = response;
     action.gcsBucket = "outputbucket";
     action.gcsBufferSize = 500;
+    action.clock = clock;
+    action.driveConnection = driveConnection;
     gcsService = createGcsService();
   }
 
-  private void runMapreduce() throws Exception {
+  private void runAction() throws Exception {
     action.run();
     executeTasksUntilEmpty("mapreduce");
   }
@@ -80,27 +96,27 @@ class ExportDomainListsActionTest extends MapreduceTestCase<ExportDomainListsAct
     verify(driveConnection)
         .createOrUpdateFile(
             eq(REGISTERED_DOMAINS_FILENAME),
-            eq(EXPORT_MIME_TYPE),
+            eq(MediaType.PLAIN_TEXT_UTF_8),
             eq(folderId),
             bytesExportedToDrive.capture());
     assertThat(new String(bytesExportedToDrive.getValue(), UTF_8)).isEqualTo(domains);
   }
 
-  @Test
+  @TestOfyOnly
   void test_writesLinkToMapreduceConsoleToResponse() throws Exception {
-    runMapreduce();
+    runAction();
     assertThat(response.getPayload())
         .startsWith(
             "Mapreduce console: https://backend-dot-projectid.appspot.com"
                 + "/_ah/pipeline/status.html?root=");
   }
 
-  @Test
+  @TestOfyAndSql
   void test_outputsOnlyActiveDomains() throws Exception {
     persistActiveDomain("onetwo.tld");
     persistActiveDomain("rudnitzky.tld");
     persistDeletedDomain("mortuary.tld", DateTime.parse("2001-03-14T10:11:12Z"));
-    runMapreduce();
+    runAction();
     GcsFilename existingFile = new GcsFilename("outputbucket", "tld.txt");
     String tlds = new String(readGcsFile(gcsService, existingFile), UTF_8);
     // Check that it only contains the active domains, not the dead one.
@@ -109,12 +125,12 @@ class ExportDomainListsActionTest extends MapreduceTestCase<ExportDomainListsAct
     verifyNoMoreInteractions(driveConnection);
   }
 
-  @Test
+  @TestOfyAndSql
   void test_outputsOnlyDomainsOnRealTlds() throws Exception {
     persistActiveDomain("onetwo.tld");
     persistActiveDomain("rudnitzky.tld");
     persistActiveDomain("wontgo.testtld");
-    runMapreduce();
+    runAction();
     GcsFilename existingFile = new GcsFilename("outputbucket", "tld.txt");
     String tlds = new String(readGcsFile(gcsService, existingFile), UTF_8).trim();
     // Check that it only contains the domains on the real TLD, and not the test one.
@@ -130,7 +146,7 @@ class ExportDomainListsActionTest extends MapreduceTestCase<ExportDomainListsAct
     verifyNoMoreInteractions(driveConnection);
   }
 
-  @Test
+  @TestOfyAndSql
   void test_outputsDomainsFromDifferentTldsToMultipleFiles() throws Exception {
     createTld("tldtwo");
     persistResource(Registry.get("tldtwo").asBuilder().setDriveFolderId("hooray").build());
@@ -143,7 +159,7 @@ class ExportDomainListsActionTest extends MapreduceTestCase<ExportDomainListsAct
     persistActiveDomain("santa.tldtwo");
     persistActiveDomain("buddy.tldtwo");
     persistActiveDomain("cupid.tldthree");
-    runMapreduce();
+    runAction();
     GcsFilename firstTldFile = new GcsFilename("outputbucket", "tld.txt");
     String tlds = new String(readGcsFile(gcsService, firstTldFile), UTF_8).trim();
     assertThat(tlds).isEqualTo("dasher.tld\nprancer.tld");

core/src/test/java/google/registry/flows/EppPointInTimeTest.java

@@ -16,8 +16,11 @@ package google.registry.flows;
 
 import static com.google.common.truth.Truth.assertThat;
 import static google.registry.model.EppResourceUtils.loadAtPointInTime;
-import static google.registry.model.ofy.ObjectifyService.auditedOfy;
+import static google.registry.model.ImmutableObjectSubject.assertAboutImmutableObjects;
+import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
 import static google.registry.testing.DatabaseHelper.createTld;
+import static google.registry.testing.DatabaseHelper.loadAllOf;
+import static google.registry.testing.DatabaseHelper.loadByEntity;
 import static google.registry.testing.DatabaseHelper.persistActiveContact;
 import static google.registry.testing.DatabaseHelper.persistActiveHost;
 import static java.nio.charset.StandardCharsets.UTF_8;
@@ -25,31 +28,38 @@ import static org.joda.time.DateTimeZone.UTC;
 import static org.joda.time.Duration.standardDays;
 
 import com.google.common.collect.ImmutableMap;
-import com.googlecode.objectify.Key;
+import com.google.common.collect.Iterables;
 import google.registry.flows.EppTestComponent.FakesAndMocksModule;
 import google.registry.model.domain.DomainBase;
 import google.registry.model.ofy.Ofy;
 import google.registry.monitoring.whitebox.EppMetric;
 import google.registry.testing.AppEngineExtension;
+import google.registry.testing.DualDatabaseTest;
 import google.registry.testing.EppLoader;
 import google.registry.testing.FakeClock;
 import google.registry.testing.FakeHttpSession;
 import google.registry.testing.InjectExtension;
+import google.registry.testing.TestOfyAndSql;
 import org.joda.time.DateTime;
 import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.RegisterExtension;
 
-/** Test that domain flows create the commit logs needed to reload at points in the past. */
-class EppCommitLogsTest {
+/** Test that we can reload EPP resources as they were in the past. */
+@DualDatabaseTest
+class EppPointInTimeTest {
+
+  private final FakeClock clock = new FakeClock(DateTime.now(UTC));
 
   @RegisterExtension
   final AppEngineExtension appEngine =
-      AppEngineExtension.builder().withDatastoreAndCloudSql().withTaskQueue().build();
+      AppEngineExtension.builder()
+          .withDatastoreAndCloudSql()
+          .withClock(clock)
+          .withTaskQueue()
+          .build();
 
   @RegisterExtension final InjectExtension inject = new InjectExtension();
 
-  private final FakeClock clock = new FakeClock(DateTime.now(UTC));
   private EppLoader eppLoader;
 
   @BeforeEach
@@ -81,7 +91,7 @@ class EppCommitLogsTest {
         .run(EppMetric.builder());
   }
 
-  @Test
+  @TestOfyAndSql
   void testLoadAtPointInTime() throws Exception {
     clock.setTo(DateTime.parse("1984-12-18T12:30Z")); // not midnight
 
@@ -95,66 +105,77 @@ class EppCommitLogsTest {
     clock.setTo(timeAtCreate);
     eppLoader = new EppLoader(this, "domain_create.xml", ImmutableMap.of("DOMAIN", "example.tld"));
     runFlow();
-    auditedOfy().clearSessionCache();
-    Key<DomainBase> key = Key.create(auditedOfy().load().type(DomainBase.class).first().now());
-    DomainBase domainAfterCreate = auditedOfy().load().key(key).now();
+    tm().clearSessionCache();
+    DomainBase domainAfterCreate = Iterables.getOnlyElement(loadAllOf(DomainBase.class));
     assertThat(domainAfterCreate.getDomainName()).isEqualTo("example.tld");
 
     clock.advanceBy(standardDays(2));
     DateTime timeAtFirstUpdate = clock.nowUtc();
     eppLoader = new EppLoader(this, "domain_update_dsdata_add.xml");
     runFlow();
-    auditedOfy().clearSessionCache();
+    tm().clearSessionCache();
 
-    DomainBase domainAfterFirstUpdate = auditedOfy().load().key(key).now();
+    DomainBase domainAfterFirstUpdate = loadByEntity(domainAfterCreate);
     assertThat(domainAfterCreate).isNotEqualTo(domainAfterFirstUpdate);
 
     clock.advanceOneMilli(); // same day as first update
     DateTime timeAtSecondUpdate = clock.nowUtc();
     eppLoader = new EppLoader(this, "domain_update_dsdata_rem.xml");
     runFlow();
-    auditedOfy().clearSessionCache();
-    DomainBase domainAfterSecondUpdate = auditedOfy().load().key(key).now();
+    tm().clearSessionCache();
+    DomainBase domainAfterSecondUpdate = loadByEntity(domainAfterCreate);
 
     clock.advanceBy(standardDays(2));
     DateTime timeAtDelete = clock.nowUtc(); // before 'add' grace period ends
     eppLoader = new EppLoader(this, "domain_delete.xml", ImmutableMap.of("DOMAIN", "example.tld"));
     runFlow();
-    auditedOfy().clearSessionCache();
+    tm().clearSessionCache();
 
     assertThat(domainAfterFirstUpdate).isNotEqualTo(domainAfterSecondUpdate);
 
     // Point-in-time can only rewind an object from the current version, not roll forward.
-    DomainBase latest = auditedOfy().load().key(key).now();
+    DomainBase latest = loadByEntity(domainAfterCreate);
 
     // Creation time has millisecond granularity due to isActive() check.
-    auditedOfy().clearSessionCache();
-    assertThat(loadAtPointInTime(latest, timeAtCreate.minusMillis(1)).now()).isNull();
-    assertThat(loadAtPointInTime(latest, timeAtCreate).now()).isNotNull();
-    assertThat(loadAtPointInTime(latest, timeAtCreate.plusMillis(1)).now()).isNotNull();
+    tm().clearSessionCache();
+    assertThat(loadAtPointInTime(latest, timeAtCreate.minusMillis(1))).isNull();
+    assertThat(loadAtPointInTime(latest, timeAtCreate)).isNotNull();
+    assertThat(loadAtPointInTime(latest, timeAtCreate.plusMillis(1))).isNotNull();
 
-    auditedOfy().clearSessionCache();
-    assertThat(loadAtPointInTime(latest, timeAtCreate.plusDays(1)).now())
-        .isEqualTo(domainAfterCreate);
+    tm().clearSessionCache();
+    assertAboutImmutableObjects()
+        .that(loadAtPointInTime(latest, timeAtCreate.plusDays(1)))
+        .hasFieldsEqualTo(domainAfterCreate);
 
-    // Both updates happened on the same day. Since the revisions field has day granularity, the
-    // key to the first update should have been overwritten by the second, and its timestamp rolled
-    // forward. So we have to fall back to the last revision before midnight.
-    auditedOfy().clearSessionCache();
-    assertThat(loadAtPointInTime(latest, timeAtFirstUpdate).now()).isEqualTo(domainAfterCreate);
+    tm().clearSessionCache();
+    if (tm().isOfy()) {
+      // Both updates happened on the same day. Since the revisions field has day granularity in
+      // Datastore, the key to the first update should have been overwritten by the second, and its
+      // timestamp rolled forward. So we have to fall back to the last revision before midnight.
+      assertThat(loadAtPointInTime(latest, timeAtFirstUpdate)).isEqualTo(domainAfterCreate);
+    } else {
+      // In SQL, however, we are not limited by the day granularity, so when we request the object
+      // at timeAtFirstUpdate we should receive the object at that first update, even though the
+      // second update occurred one millisecond later.
+      assertAboutImmutableObjects()
+          .that(loadAtPointInTime(latest, timeAtFirstUpdate))
+          .hasFieldsEqualTo(domainAfterFirstUpdate);
+    }
 
-    auditedOfy().clearSessionCache();
-    assertThat(loadAtPointInTime(latest, timeAtSecondUpdate).now())
-        .isEqualTo(domainAfterSecondUpdate);
+    tm().clearSessionCache();
+    assertAboutImmutableObjects()
+        .that(loadAtPointInTime(latest, timeAtSecondUpdate))
+        .hasFieldsEqualTo(domainAfterSecondUpdate);
 
-    auditedOfy().clearSessionCache();
-    assertThat(loadAtPointInTime(latest, timeAtSecondUpdate.plusDays(1)).now())
-        .isEqualTo(domainAfterSecondUpdate);
+    tm().clearSessionCache();
+    assertAboutImmutableObjects()
+        .that(loadAtPointInTime(latest, timeAtSecondUpdate.plusDays(1)))
+        .hasFieldsEqualTo(domainAfterSecondUpdate);
 
     // Deletion time has millisecond granularity due to isActive() check.
-    auditedOfy().clearSessionCache();
-    assertThat(loadAtPointInTime(latest, timeAtDelete.minusMillis(1)).now()).isNotNull();
-    assertThat(loadAtPointInTime(latest, timeAtDelete).now()).isNull();
-    assertThat(loadAtPointInTime(latest, timeAtDelete.plusMillis(1)).now()).isNull();
+    tm().clearSessionCache();
+    assertThat(loadAtPointInTime(latest, timeAtDelete.minusMillis(1))).isNotNull();
+    assertThat(loadAtPointInTime(latest, timeAtDelete)).isNull();
+    assertThat(loadAtPointInTime(latest, timeAtDelete.plusMillis(1))).isNull();
   }
 }

core/src/test/java/google/registry/model/EppResourceUtilsTest.java

@@ -41,14 +41,18 @@ import org.junit.jupiter.api.extension.RegisterExtension;
 @DualDatabaseTest
 class EppResourceUtilsTest {
 
+  private final FakeClock clock = new FakeClock(DateTime.now(UTC));
+
   @RegisterExtension
   public final AppEngineExtension appEngine =
-      AppEngineExtension.builder().withDatastoreAndCloudSql().withTaskQueue().build();
+      AppEngineExtension.builder()
+          .withDatastoreAndCloudSql()
+          .withClock(clock)
+          .withTaskQueue()
+          .build();
 
   @RegisterExtension public final InjectExtension inject = new InjectExtension();
 
-  private final FakeClock clock = new FakeClock(DateTime.now(UTC));
-
   @BeforeEach
   void beforeEach() {
     createTld("tld");
@@ -63,7 +67,7 @@ class EppResourceUtilsTest {
         newHostResource("ns1.cat.tld").asBuilder()
             .setCreationTimeForTest(clock.nowUtc())
             .build());
-    assertThat(loadAtPointInTime(host, clock.nowUtc().minus(Duration.millis(1))).now()).isNull();
+    assertThat(loadAtPointInTime(host, clock.nowUtc().minus(Duration.millis(1)))).isNull();
   }
 
   @TestOfyAndSql
@@ -74,7 +78,7 @@ class EppResourceUtilsTest {
         newHostResource("ns1.cat.tld").asBuilder()
             .setCreationTimeForTest(START_OF_TIME)
             .build());
-    assertThat(loadAtPointInTime(host, clock.nowUtc()).now()).isEqualTo(host);
+    assertThat(loadAtPointInTime(host, clock.nowUtc())).isEqualTo(host);
   }
 
   @TestOfyOnly
@@ -95,8 +99,7 @@ class EppResourceUtilsTest {
             .build());
     // Load at the point in time just before the latest update; the floor entry of the revisions
     // map should point to the manifest for the first save, so we should get the old host.
-    assertThat(loadAtPointInTime(currentHost, clock.nowUtc().minusMillis(1)).now())
-        .isEqualTo(oldHost);
+    assertThat(loadAtPointInTime(currentHost, clock.nowUtc().minusMillis(1))).isEqualTo(oldHost);
   }
 
   @TestOfyOnly
@@ -116,7 +119,7 @@ class EppResourceUtilsTest {
     // Load at the point in time just before the latest update; the old host is not recoverable
     // (revisions map link is broken, and guessing using the oldest revision map entry finds the
     // same broken link), so just returns the current host.
-    assertThat(loadAtPointInTime(host, clock.nowUtc().minusMillis(1)).now()).isEqualTo(host);
+    assertThat(loadAtPointInTime(host, clock.nowUtc().minusMillis(1))).isEqualTo(host);
   }
 
   @TestOfyOnly
@@ -137,8 +140,7 @@ class EppResourceUtilsTest {
     // Load at the point in time before the first update; there will be no floor entry for the
     // revisions map, so give up and return the oldest revision entry's mutation value (the old host
     // data).
-    assertThat(loadAtPointInTime(currentHost, clock.nowUtc().minusDays(2)).now())
-        .isEqualTo(oldHost);
+    assertThat(loadAtPointInTime(currentHost, clock.nowUtc().minusDays(2))).isEqualTo(oldHost);
   }
 
   @TestOfyOnly
@@ -153,7 +155,7 @@ class EppResourceUtilsTest {
     // Load at the point in time before the first save; there will be no floor entry for the
     // revisions map.  Since the oldest revision entry is the only (i.e. current) revision, return
     // the resource.
-    assertThat(loadAtPointInTime(host, clock.nowUtc().minusMillis(1)).now()).isEqualTo(host);
+    assertThat(loadAtPointInTime(host, clock.nowUtc().minusMillis(1))).isEqualTo(host);
   }
 
   @TestOfyOnly
@@ -171,7 +173,7 @@ class EppResourceUtilsTest {
     // Even though there is no revision, make a best effort guess to use the oldest revision.
     assertThat(
             loadAtPointInTime(host, clock.nowUtc().minus(Duration.standardDays(32)))
-                .now()
+
                 .getUpdateTimestamp()
                 .getTimestamp())
         .isEqualTo(host.getRevisions().firstKey());

core/src/test/java/google/registry/reporting/spec11/GenerateSpec11ReportActionTest.java

@@ -50,6 +50,7 @@ class GenerateSpec11ReportActionTest extends BeamActionTestBase {
             "gs://reporting-project/reporting-bucket/",
             "api_key/a",
             clock.nowUtc().toLocalDate(),
+            "DATASTORE",
             clock,
             response,
             dataflow);
@@ -71,6 +72,7 @@ class GenerateSpec11ReportActionTest extends BeamActionTestBase {
             "gs://reporting-project/reporting-bucket/",
             "api_key/a",
             clock.nowUtc().toLocalDate(),
+            "DATASTORE",
             clock,
             response,
             dataflow);

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

@@ -20,6 +20,7 @@ import static google.registry.persistence.transaction.TransactionManagerFactory.
 import static google.registry.testing.LogsSubject.assertAboutLogs;
 import static google.registry.util.DateTimeUtils.START_OF_TIME;
 
+import com.google.common.base.Suppliers;
 import com.google.common.collect.ImmutableSortedMap;
 import com.google.common.testing.TestLogHandler;
 import com.googlecode.objectify.Key;
@@ -29,6 +30,7 @@ import google.registry.config.RegistryConfig;
 import google.registry.model.ImmutableObject;
 import google.registry.model.common.DatabaseMigrationStateSchedule;
 import google.registry.model.common.DatabaseMigrationStateSchedule.MigrationState;
+import google.registry.model.ofy.CommitLogBucket;
 import google.registry.model.ofy.Ofy;
 import google.registry.persistence.VKey;
 import google.registry.persistence.transaction.TransactionEntity;
@@ -66,6 +68,10 @@ public class ReplicateToDatastoreActionTest {
   @BeforeEach
   public void setUp() {
     injectExtension.setStaticField(Ofy.class, "clock", fakeClock);
+    // Use a single bucket to expose timestamp inversion problems.
+    injectExtension.setStaticField(
+        CommitLogBucket.class, "bucketIdSupplier", Suppliers.ofInstance(1));
+    fakeClock.setAutoIncrementByOneMilli();
     RegistryConfig.overrideCloudSqlReplicateTransactions(true);
     Logger.getLogger(ReplicateToDatastoreAction.class.getCanonicalName()).addHandler(logHandler);
     DateTime now = fakeClock.nowUtc();
@@ -87,6 +93,7 @@ public class ReplicateToDatastoreActionTest {
 
   @AfterEach
   public void tearDown() {
+    fakeClock.disableAutoIncrement();
     RegistryConfig.overrideCloudSqlReplicateTransactions(false);
   }
 

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

@@ -162,6 +162,20 @@ class CreateDomainCommandTest extends EppToolCommandTestCase<CreateDomainCommand
             + "sending total cost for 1 year(s) of USD 877.00.");
   }
 
+  @Test
+  void testSuccess_reasonAndRegistrarRequest() throws Exception {
+    createTld("tld");
+    runCommandForced(
+        "--client=NewRegistrar",
+        "--registrant=crr-admin",
+        "--admins=crr-admin",
+        "--techs=crr-tech",
+        "--reason=\"Creating test domain\"",
+        "--registrar_request=false",
+        "example.tld");
+    eppVerifier.verifySent("domain_create_metadata.xml");
+  }
+
   @Test
   void testFailure_duplicateDomains() {
     IllegalArgumentException thrown =
@@ -175,7 +189,7 @@ class CreateDomainCommandTest extends EppToolCommandTestCase<CreateDomainCommand
                     "--techs=crr-tech",
                     "example.tld",
                     "example.tld"));
-    assertThat(thrown).hasMessageThat().contains("Duplicate arguments found: \'example.tld\'");
+    assertThat(thrown).hasMessageThat().contains("Duplicate arguments found: 'example.tld'");
   }
 
   @Test

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

@@ -197,6 +197,18 @@ class UpdateDomainCommandTest extends EppToolCommandTestCase<UpdateDomainCommand
     eppVerifier.verifySent("domain_update_change.xml");
   }
 
+  @TestOfyAndSql
+  void testSuccess_change_reasonAndRegistrarRequest() throws Exception {
+    runCommandForced(
+        "--client=NewRegistrar",
+        "--registrant=crr-admin",
+        "--password=2fooBAR",
+        "--reason=\"Testing domain update\"",
+        "--registrar_request=false",
+        "example.tld");
+    eppVerifier.verifySent("domain_update_change_metadata.xml");
+  }
+
   @TestOfyAndSql
   void testSuccess_setNameservers() throws Exception {
     HostResource host1 = persistActiveHost("ns1.zdns.google");

core/src/test/java/google/registry/tools/server/RefreshDnsForAllDomainsActionTest.java

@@ -18,37 +18,52 @@ import static com.google.common.truth.Truth.assertThat;
 import static google.registry.testing.DatabaseHelper.createTld;
 import static google.registry.testing.DatabaseHelper.persistActiveDomain;
 import static google.registry.testing.DatabaseHelper.persistDeletedDomain;
-import static org.joda.time.DateTimeZone.UTC;
+import static google.registry.testing.TaskQueueHelper.assertDnsTasksEnqueued;
 import static org.joda.time.Duration.standardMinutes;
 import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.verify;
 
 import com.google.common.collect.ImmutableSet;
 import google.registry.dns.DnsQueue;
+import google.registry.model.ofy.Ofy;
+import google.registry.testing.DualDatabaseTest;
+import google.registry.testing.FakeClock;
 import google.registry.testing.FakeResponse;
 import google.registry.testing.InjectExtension;
+import google.registry.testing.TestOfyAndSql;
+import google.registry.testing.TestSqlOnly;
 import google.registry.testing.mapreduce.MapreduceTestCase;
 import google.registry.tools.server.RefreshDnsForAllDomainsAction.RefreshDnsForAllDomainsActionMapper;
 import java.util.Random;
+import org.apache.http.HttpStatus;
 import org.joda.time.DateTime;
 import org.joda.time.Duration;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Order;
 import org.junit.jupiter.api.extension.RegisterExtension;
 import org.mockito.ArgumentCaptor;
 
 /** Unit tests for {@link RefreshDnsForAllDomainsAction}. */
+@DualDatabaseTest
 public class RefreshDnsForAllDomainsActionTest
     extends MapreduceTestCase<RefreshDnsForAllDomainsAction> {
 
-  @RegisterExtension public final InjectExtension inject = new InjectExtension();
-
+  private final FakeClock clock = new FakeClock(DateTime.parse("2020-02-02T02:02:02Z"));
   private final DnsQueue dnsQueue = mock(DnsQueue.class);
   private DnsQueue origDnsQueue;
+  private FakeResponse response = new FakeResponse();
+
+  @Order(Order.DEFAULT - 1)
+  @RegisterExtension
+  public final InjectExtension inject =
+      new InjectExtension().withStaticFieldOverride(Ofy.class, "clock", clock);
 
   @BeforeEach
   void beforeEach() {
@@ -60,6 +75,9 @@ public class RefreshDnsForAllDomainsActionTest
     action.random.setSeed(123L);
     action.mrRunner = makeDefaultRunner();
     action.response = new FakeResponse();
+    action.clock = clock;
+    action.dnsQueue = dnsQueue;
+    action.response = response;
 
     createTld("bar");
   }
@@ -70,58 +88,74 @@ public class RefreshDnsForAllDomainsActionTest
         .isEqualTo(dnsQueue);
   }
 
-  private void runMapreduce() throws Exception {
+  private void runAction() throws Exception {
     action.run();
     executeTasksUntilEmpty("mapreduce");
   }
 
-  @Test
+  @TestSqlOnly
+  void test_runAction_errorEnqueuingToDnsQueue() throws Exception {
+    persistActiveDomain("foo.bar");
+    persistActiveDomain("baz.bar");
+    persistActiveDomain("low.bar");
+    action.tlds = ImmutableSet.of("bar");
+    DnsQueue faultyQueue = spy(origDnsQueue);
+    doThrow(new RuntimeException("Error enqueuing task."))
+        .when(faultyQueue)
+        .addDomainRefreshTask(eq("baz.bar"), any(Duration.class));
+    action.dnsQueue = faultyQueue;
+    runAction();
+    assertDnsTasksEnqueued("foo.bar", "low.bar");
+    assertThat(response.getStatus()).isEqualTo(HttpStatus.SC_INTERNAL_SERVER_ERROR);
+  }
+
+  @TestOfyAndSql
   void test_runAction_successfullyEnqueuesDnsRefreshes() throws Exception {
     persistActiveDomain("foo.bar");
     persistActiveDomain("low.bar");
     action.tlds = ImmutableSet.of("bar");
-    runMapreduce();
+    runAction();
     verify(dnsQueue).addDomainRefreshTask("foo.bar", Duration.ZERO);
     verify(dnsQueue).addDomainRefreshTask("low.bar", Duration.ZERO);
   }
 
-  @Test
+  @TestOfyAndSql
   void test_runAction_smearsOutDnsRefreshes() throws Exception {
     persistActiveDomain("foo.bar");
     persistActiveDomain("low.bar");
     action.tlds = ImmutableSet.of("bar");
     action.smearMinutes = 1000;
-    runMapreduce();
+    runAction();
     ArgumentCaptor<Duration> captor = ArgumentCaptor.forClass(Duration.class);
     verify(dnsQueue).addDomainRefreshTask(eq("foo.bar"), captor.capture());
     verify(dnsQueue).addDomainRefreshTask(eq("low.bar"), captor.capture());
     assertThat(captor.getAllValues()).containsExactly(standardMinutes(450), standardMinutes(782));
   }
 
-  @Test
+  @TestOfyAndSql
   void test_runAction_doesntRefreshDeletedDomain() throws Exception {
     persistActiveDomain("foo.bar");
-    persistDeletedDomain("deleted.bar", DateTime.now(UTC).minusYears(1));
+    persistDeletedDomain("deleted.bar", clock.nowUtc().minusYears(1));
     action.tlds = ImmutableSet.of("bar");
-    runMapreduce();
+    runAction();
     verify(dnsQueue).addDomainRefreshTask("foo.bar", Duration.ZERO);
     verify(dnsQueue, never()).addDomainRefreshTask("deleted.bar", Duration.ZERO);
   }
 
-  @Test
+  @TestOfyAndSql
   void test_runAction_ignoresDomainsOnOtherTlds() throws Exception {
     createTld("baz");
     persistActiveDomain("foo.bar");
     persistActiveDomain("low.bar");
     persistActiveDomain("ignore.baz");
     action.tlds = ImmutableSet.of("bar");
-    runMapreduce();
+    runAction();
     verify(dnsQueue).addDomainRefreshTask("foo.bar", Duration.ZERO);
     verify(dnsQueue).addDomainRefreshTask("low.bar", Duration.ZERO);
     verify(dnsQueue, never()).addDomainRefreshTask("ignore.baz", Duration.ZERO);
   }
 
-  @Test
+  @TestOfyAndSql
   void test_smearMinutesMustBeSpecified() {
     action.tlds = ImmutableSet.of("bar");
     action.smearMinutes = 0;

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

@@ -667,6 +667,7 @@ class google.registry.model.registry.Registry {
   int numDnsPublishLocks;
   java.lang.String driveFolderId;
   java.lang.String lordnUsername;
+  java.lang.String premiumListName;
   java.lang.String pricingEngineClassName;
   java.lang.String roidSuffix;
   java.lang.String tldStr;
@@ -675,6 +676,7 @@ class google.registry.model.registry.Registry {
   java.util.Set<java.lang.String> allowedFullyQualifiedHostNames;
   java.util.Set<java.lang.String> allowedRegistrantContactIds;
   java.util.Set<java.lang.String> dnsWriters;
+  java.util.Set<java.lang.String> reservedListNames;
   org.joda.money.CurrencyUnit currency;
   org.joda.money.Money createBillingCost;
   org.joda.money.Money registryLockOrUnlockBillingCost;

core/src/test/resources/google/registry/module/backend/backend_routing.txt

@@ -34,6 +34,7 @@ PATH                                    CLASS                               METH
 /_dr/task/rdeUpload                     RdeUploadAction                     POST     n  INTERNAL,API APP ADMIN
 /_dr/task/refreshDnsOnHostRename        RefreshDnsOnHostRenameAction        GET      n  INTERNAL,API APP ADMIN
 /_dr/task/relockDomain                  RelockDomainAction                  POST     y  INTERNAL,API APP ADMIN
+/_dr/task/replayCommitLogsToSql         ReplayCommitLogsToSqlAction         POST     y  INTERNAL,API APP ADMIN
 /_dr/task/resaveAllEppResources         ResaveAllEppResourcesAction         GET      n  INTERNAL,API APP ADMIN
 /_dr/task/resaveEntity                  ResaveEntityAction                  POST     n  INTERNAL,API APP ADMIN
 /_dr/task/syncGroupMembers              SyncGroupMembersAction              POST     n  INTERNAL,API APP ADMIN

core/src/test/resources/google/registry/tools/server/domain_create_metadata.xml

@@ -0,0 +1,25 @@
+<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
+<epp xmlns="urn:ietf:params:xml:ns:epp-1.0">
+  <command>
+    <create>
+      <domain:create
+       xmlns:domain="urn:ietf:params:xml:ns:domain-1.0">
+        <domain:name>example.tld</domain:name>
+        <domain:period unit="y">1</domain:period>
+        <domain:registrant>crr-admin</domain:registrant>
+        <domain:contact type="admin">crr-admin</domain:contact>
+        <domain:contact type="tech">crr-tech</domain:contact>
+        <domain:authInfo>
+          <domain:pw>abcdefghijklmnop</domain:pw>
+        </domain:authInfo>
+      </domain:create>
+    </create>
+    <extension>
+      <metadata:metadata xmlns:metadata="urn:google:params:xml:ns:metadata-1.0">
+        <metadata:reason>Creating test domain</metadata:reason>
+        <metadata:requestedByRegistrar>false</metadata:requestedByRegistrar>
+      </metadata:metadata>
+    </extension>
+    <clTRID>RegistryTool</clTRID>
+  </command>
+</epp>

core/src/test/resources/google/registry/tools/server/domain_update_change_metadata.xml

@@ -0,0 +1,24 @@
+<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
+<epp xmlns="urn:ietf:params:xml:ns:epp-1.0">
+  <command>
+    <update>
+      <domain:update
+       xmlns:domain="urn:ietf:params:xml:ns:domain-1.0">
+        <domain:name>example.tld</domain:name>
+        <domain:chg>
+          <domain:registrant>crr-admin</domain:registrant>
+          <domain:authInfo>
+            <domain:pw>2fooBAR</domain:pw>
+          </domain:authInfo>
+        </domain:chg>
+      </domain:update>
+    </update>
+    <extension>
+      <metadata:metadata xmlns:metadata="urn:google:params:xml:ns:metadata-1.0">
+        <metadata:reason>Testing domain update</metadata:reason>
+        <metadata:requestedByRegistrar>false</metadata:requestedByRegistrar>
+      </metadata:metadata>
+    </extension>
+    <clTRID>RegistryTool</clTRID>
+  </command>
+</epp>