Add a dry-run option to commit-log-replay action and use it in Sandbox (#1234 )

Add more logging to replay commit logs action (#1232 )
Set payload response in happy path of ReplayCommitLogsToSqlAction (#1229 )
2026-07-10 18:13:02 +00:00 · 2021-07-03 11:18:00 -04:00 · 2021-07-02 18:06:04 -04:00 · 2021-07-01 18:21:17 -04:00 · 2021-06-30 13:54:24 -04:00 · 2021-06-29 17:12:02 -04:00
64 changed files with 1225 additions and 262 deletions
@@ -79,6 +79,9 @@ def fragileTestPatterns = [
    // Changes cache timeouts and for some reason appears to have contention
    // with other tests.
    "google/registry/whois/WhoisCommandFactoryTest.*",
+    // Currently changes a global configuration parameter that for some reason
+    // results in timestamp inversions for other tests.  TODO(mmuller): fix.
+    "google/registry/flows/host/HostInfoFlowTest.*",
 ] + dockerIncompatibleTestPatterns

 sourceSets {
@@ -777,6 +780,10 @@ if (environment in ['alpha', 'crash']) {
          mainClass: 'google.registry.beam.invoicing.InvoicingPipeline',
          metaData: 'google/registry/beam/invoicing_pipeline_metadata.json'
      ],
+      [
+          mainClass: 'google.registry.beam.rde.RdePipeline',
+          metaData: 'google/registry/beam/rde_pipeline_metadata.json'
+      ],
  ]
  project.tasks.create("stage_beam_pipelines") {
    doLast {
@@ -14,7 +14,9 @@

 package google.registry.backup;

+import static com.google.common.collect.ImmutableList.toImmutableList;
 import static google.registry.backup.ExportCommitLogDiffAction.DIFF_FILE_PREFIX;
+import static google.registry.backup.RestoreCommitLogsAction.DRY_RUN_PARAM;
 import static google.registry.model.ofy.EntityWritePriorities.getEntityPriority;
 import static google.registry.model.ofy.ObjectifyService.auditedOfy;
 import static google.registry.persistence.transaction.TransactionManagerFactory.jpaTm;
@@ -25,6 +27,7 @@ import com.google.appengine.api.datastore.Entity;
 import com.google.appengine.api.datastore.Key;
 import com.google.appengine.tools.cloudstorage.GcsFileMetadata;
 import com.google.appengine.tools.cloudstorage.GcsService;
+import com.google.common.base.Joiner;
 import com.google.common.collect.ImmutableList;
 import com.google.common.flogger.FluentLogger;
 import google.registry.model.common.DatabaseMigrationStateSchedule;
@@ -35,6 +38,7 @@ 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.Parameter;
 import google.registry.request.Response;
 import google.registry.request.auth.Auth;
 import google.registry.schema.replay.DatastoreEntity;
@@ -64,10 +68,13 @@ public class ReplayCommitLogsToSqlAction implements Runnable {

  static final String PATH = "/_dr/task/replayCommitLogsToSql";

+  private static final FluentLogger logger = FluentLogger.forEnclosingClass();
  private static final int BLOCK_SIZE =
      1024 * 1024; // Buffer 1mb at a time, for no particular reason.
-  private static final FluentLogger logger = FluentLogger.forEnclosingClass();
  private static final Duration LEASE_LENGTH = standardHours(1);
+  // Stop / pause where we are if we've been replaying for more than five minutes to avoid GAE
+  // request timeouts
+  private static final Duration REPLAY_TIMEOUT_DURATION = Duration.standardMinutes(5);

  @Inject GcsService gcsService;
  @Inject Response response;
@@ -75,6 +82,11 @@ public class ReplayCommitLogsToSqlAction implements Runnable {
  @Inject GcsDiffFileLister diffLister;
  @Inject Clock clock;

+  /** If true, will exit after logging the commit log files that would otherwise be replayed. */
+  @Inject
+  @Parameter(DRY_RUN_PARAM)
+  boolean dryRun;
+
  @Inject
  ReplayCommitLogsToSqlAction() {}

@@ -104,44 +116,91 @@ public class ReplayCommitLogsToSqlAction implements Runnable {
      return;
    }
    try {
-      replayFiles();
-      response.setStatus(HttpServletResponse.SC_OK);
-      logger.atInfo().log("ReplayCommitLogsToSqlAction completed successfully.");
+      logger.atInfo().log("Beginning replay of commit logs.");
+      ImmutableList<GcsFileMetadata> commitLogFiles = getFilesToReplay();
+      if (dryRun) {
+        response.setStatus(HttpServletResponse.SC_OK);
+        ImmutableList<String> filenames =
+            commitLogFiles.stream()
+                .limit(10)
+                .map(file -> file.getFilename().getObjectName())
+                .collect(toImmutableList());
+        String dryRunMessage =
+            "Running in dry-run mode; would have processed %d files. They are (limit 10):\n"
+                + Joiner.on('\n').join(filenames);
+        response.setPayload(dryRunMessage);
+        logger.atInfo().log(dryRunMessage);
+      } else {
+        replayFiles(commitLogFiles);
+        response.setStatus(HttpServletResponse.SC_OK);
+        String message = "ReplayCommitLogsToSqlAction completed successfully.";
+        response.setPayload(message);
+        logger.atInfo().log(message);
+      }
+    } catch (Throwable t) {
+      String message = "Errored out replaying files.";
+      logger.atSevere().withCause(t).log(message);
+      response.setStatus(HttpServletResponse.SC_INTERNAL_SERVER_ERROR);
+      response.setPayload(message);
    } finally {
      lock.ifPresent(Lock::release);
    }
  }

-  private void replayFiles() {
+  private ImmutableList<GcsFileMetadata> getFilesToReplay() {
    // Start at the first millisecond we haven't seen yet
    DateTime fromTime = jpaTm().transact(() -> SqlReplayCheckpoint.get().plusMillis(1));
+    logger.atInfo().log("Starting replay from: %s.", fromTime);
    // If there's an inconsistent file set, this will throw IllegalStateException and the job
    // will try later -- this is likely because an export hasn't finished yet.
    ImmutableList<GcsFileMetadata> commitLogFiles =
        diffLister.listDiffFiles(fromTime, /* current time */ null);
+    logger.atInfo().log("Found %d new commit log files to process.", commitLogFiles.size());
+    return commitLogFiles;
+  }
+
+  private void replayFiles(ImmutableList<GcsFileMetadata> commitLogFiles) {
+    DateTime replayTimeoutTime = clock.nowUtc().plus(REPLAY_TIMEOUT_DURATION);
+    int processedFiles = 0;
    for (GcsFileMetadata metadata : commitLogFiles) {
      // One transaction per GCS file
      jpaTm().transact(() -> processFile(metadata));
+      processedFiles++;
+      if (clock.nowUtc().isAfter(replayTimeoutTime)) {
+        logger.atInfo().log(
+            "Reached max execution time after replaying %d files, leaving %d files for next run.",
+            processedFiles, commitLogFiles.size() - processedFiles);
+        return;
+      }
    }
-    logger.atInfo().log("Replayed %d commit log files to SQL successfully.", commitLogFiles.size());
+    logger.atInfo().log("Replayed %d commit log files to SQL successfully.", processedFiles);
  }

  private void processFile(GcsFileMetadata metadata) {
+    logger.atInfo().log(
+        "Processing commit log file %s of size %d B.",
+        metadata.getFilename(), metadata.getLength());
    try (InputStream input =
        Channels.newInputStream(
            gcsService.openPrefetchingReadChannel(metadata.getFilename(), 0, BLOCK_SIZE))) {
      // Load and process the Datastore transactions one at a time
      ImmutableList<ImmutableList<VersionedEntity>> allTransactions =
          CommitLogImports.loadEntitiesByTransaction(input);
+      logger.atInfo().log(
+          "Replaying %d transactions from commit log file %s.",
+          allTransactions.size(), metadata.getFilename());
      allTransactions.forEach(this::replayTransaction);
      // if we succeeded, set the last-seen time
      DateTime checkpoint =
          DateTime.parse(
              metadata.getFilename().getObjectName().substring(DIFF_FILE_PREFIX.length()));
      SqlReplayCheckpoint.set(checkpoint);
-      logger.atInfo().log("Replayed %d transactions from commit log file.", allTransactions.size());
+      logger.atInfo().log(
+          "Replayed %d transactions from commit log file %s.",
+          allTransactions.size(), metadata.getFilename());
    } catch (IOException e) {
-      throw new RuntimeException(e);
+      throw new RuntimeException(
+          "Errored out while replaying commit log file " + metadata.getFilename(), e);
    }
  }

@@ -0,0 +1,260 @@
+// 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.beam.rde;
+
+import static com.google.common.collect.ImmutableSet.toImmutableSet;
+import static google.registry.model.EppResourceUtils.loadAtPointInTimeAsync;
+import static google.registry.persistence.transaction.TransactionManagerFactory.jpaTm;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.ImmutableMap;
+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.google.common.io.BaseEncoding;
+import google.registry.beam.common.RegistryJpaIO;
+import google.registry.model.EppResource;
+import google.registry.model.contact.ContactResource;
+import google.registry.model.domain.DomainBase;
+import google.registry.model.host.HostResource;
+import google.registry.model.rde.RdeMode;
+import google.registry.model.registrar.Registrar;
+import google.registry.model.registrar.Registrar.Type;
+import google.registry.persistence.VKey;
+import google.registry.rde.DepositFragment;
+import google.registry.rde.PendingDeposit;
+import google.registry.rde.PendingDeposit.PendingDepositCoder;
+import google.registry.rde.RdeFragmenter;
+import google.registry.rde.RdeMarshaller;
+import google.registry.xml.ValidationMode;
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+import java.util.function.Supplier;
+import javax.persistence.Entity;
+import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.PipelineResult;
+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.transforms.FlatMapElements;
+import org.apache.beam.sdk.transforms.Flatten;
+import org.apache.beam.sdk.transforms.Reshuffle;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PCollectionList;
+import org.apache.beam.sdk.values.TypeDescriptor;
+import org.apache.beam.sdk.values.TypeDescriptors;
+import org.joda.time.DateTime;
+
+/**
+ * Definition of a Dataflow Flex template, which generates RDE/BRDA deposits.
+ *
+ * <p>To stage this template locally, run the {@code stage_beam_pipeline.sh} shell script.
+ *
+ * <p>Then, you can run the staged template via the API client library, gCloud or a raw REST call.
+ *
+ * @see <a href="https://cloud.google.com/dataflow/docs/guides/templates/using-flex-templates">Using
+ *     Flex Templates</a>
+ */
+public class RdePipeline implements Serializable {
+
+  private final RdeMarshaller marshaller;
+  private final ImmutableSetMultimap<String, PendingDeposit> pendings;
+
+  // Registrars to be excluded from data escrow. Not including the sandbox-only OTE type so that
+  // if sneaks into production we would get an extra signal.
+  private static final ImmutableSet<Type> IGNORED_REGISTRAR_TYPES =
+      Sets.immutableEnumSet(Registrar.Type.MONITORING, Registrar.Type.TEST);
+
+  private static final String EPP_RESOURCE_QUERY =
+      "SELECT id FROM %entity% "
+          + "WHERE COALESCE(creationClientId, '') NOT LIKE 'prober-%' "
+          + "AND COALESCE(currentSponsorClientId, '') NOT LIKE 'prober-%' "
+          + "AND COALESCE(lastEppUpdateClientId, '') NOT LIKE 'prober-%'";
+
+  public static String createEppResourceQuery(Class<? extends EppResource> clazz) {
+    return EPP_RESOURCE_QUERY.replace("%entity%", clazz.getAnnotation(Entity.class).name())
+        + (clazz.equals(DomainBase.class) ? " AND tld in (:tlds)" : "");
+  }
+
+  RdePipeline(RdePipelineOptions options) throws IOException, ClassNotFoundException {
+    this.marshaller = new RdeMarshaller(ValidationMode.valueOf(options.getValidationMode()));
+    this.pendings = decodePendings(options.getPendings());
+  }
+
+  @VisibleForTesting
+  PipelineResult run(Pipeline pipeline) {
+    createFragments(pipeline);
+    return pipeline.run();
+  }
+
+  PipelineResult run() {
+    return run(Pipeline.create());
+  }
+
+  PCollection<KV<PendingDeposit, DepositFragment>> createFragments(Pipeline pipeline) {
+    PCollection<KV<PendingDeposit, DepositFragment>> fragments =
+        PCollectionList.of(processRegistrars(pipeline))
+            .and(processNonRegistrarEntities(pipeline, DomainBase.class))
+            .and(processNonRegistrarEntities(pipeline, ContactResource.class))
+            .and(processNonRegistrarEntities(pipeline, HostResource.class))
+            .apply(Flatten.pCollections())
+            .setCoder(
+                KvCoder.of(PendingDepositCoder.of(), SerializableCoder.of(DepositFragment.class)));
+    return fragments;
+  }
+
+  PCollection<KV<PendingDeposit, DepositFragment>> processRegistrars(Pipeline pipeline) {
+    return pipeline
+        .apply(
+            "Read all production Registrar entities",
+            RegistryJpaIO.read(
+                "SELECT clientIdentifier FROM Registrar WHERE type NOT IN (:types)",
+                ImmutableMap.of("types", IGNORED_REGISTRAR_TYPES),
+                String.class,
+                // TODO: consider adding coders for entities and pass them directly instead of using
+                // VKeys.
+                id -> VKey.createSql(Registrar.class, id)))
+        .apply(
+            "Marshal Registrar into DepositFragment",
+            FlatMapElements.into(
+                    TypeDescriptors.kvs(
+                        TypeDescriptor.of(PendingDeposit.class),
+                        TypeDescriptor.of(DepositFragment.class)))
+                .via(
+                    (VKey<Registrar> key) -> {
+                      Registrar registrar = jpaTm().transact(() -> jpaTm().loadByKey(key));
+                      DepositFragment fragment = marshaller.marshalRegistrar(registrar);
+                      return pendings.values().stream()
+                          .map(pending -> KV.of(pending, fragment))
+                          .collect(toImmutableSet());
+                    }));
+  }
+
+  @SuppressWarnings("deprecation") // Reshuffle is still recommended by Dataflow.
+  <T extends EppResource>
+      PCollection<KV<PendingDeposit, DepositFragment>> processNonRegistrarEntities(
+          Pipeline pipeline, Class<T> clazz) {
+    return createInputs(pipeline, clazz)
+        .apply("Marshal " + clazz.getSimpleName() + " into DepositFragment", mapToFragments(clazz))
+        .setCoder(KvCoder.of(PendingDepositCoder.of(), SerializableCoder.of(DepositFragment.class)))
+        .apply(
+            "Reshuffle KV<PendingDeposit, DepositFragment> of "
+                + clazz.getSimpleName()
+                + " to prevent fusion",
+            Reshuffle.of());
+  }
+
+  <T extends EppResource> PCollection<VKey<T>> createInputs(Pipeline pipeline, Class<T> clazz) {
+    return pipeline.apply(
+        "Read all production " + clazz.getSimpleName() + " entities",
+        RegistryJpaIO.read(
+            createEppResourceQuery(clazz),
+            clazz.equals(DomainBase.class)
+                ? ImmutableMap.of("tlds", pendings.keySet())
+                : ImmutableMap.of(),
+            String.class,
+            // TODO: consider adding coders for entities and pass them directly instead of using
+            // VKeys.
+            x -> VKey.create(clazz, x)));
+  }
+
+  <T extends EppResource>
+      FlatMapElements<VKey<T>, KV<PendingDeposit, DepositFragment>> mapToFragments(Class<T> clazz) {
+    return FlatMapElements.into(
+            TypeDescriptors.kvs(
+                TypeDescriptor.of(PendingDeposit.class), TypeDescriptor.of(DepositFragment.class)))
+        .via(
+            (VKey<T> key) -> {
+              T resource = jpaTm().transact(() -> jpaTm().loadByKey(key));
+              // The set of all TLDs to which this resource should be emitted.
+              ImmutableSet<String> tlds =
+                  clazz.equals(DomainBase.class)
+                      ? ImmutableSet.of(((DomainBase) resource).getTld())
+                      : pendings.keySet();
+              // Get the set of all point-in-time watermarks we need, to minimize rewinding.
+              ImmutableSet<DateTime> dates =
+                  tlds.stream()
+                      .map(pendings::get)
+                      .flatMap(ImmutableSet::stream)
+                      .map(PendingDeposit::watermark)
+                      .collect(toImmutableSet());
+              // Launch asynchronous fetches of point-in-time representations of resource.
+              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.
+              RdeFragmenter fragmenter = new RdeFragmenter(resourceAtTimes, marshaller);
+              List<KV<PendingDeposit, DepositFragment>> results = new ArrayList<>();
+              for (String tld : tlds) {
+                for (PendingDeposit pending : pendings.get(tld)) {
+                  // Hosts and contacts don't get included in BRDA deposits.
+                  if (pending.mode() == RdeMode.THIN && !clazz.equals(DomainBase.class)) {
+                    continue;
+                  }
+                  Optional<DepositFragment> fragment =
+                      fragmenter.marshal(pending.watermark(), pending.mode());
+                  fragment.ifPresent(
+                      depositFragment -> results.add(KV.of(pending, depositFragment)));
+                }
+              }
+              return results;
+            });
+  }
+
+  /**
+   * Decodes the pipeline option extracted from the URL parameter sent by the pipeline launcher to
+   * the original TLD to pending deposit map.
+   */
+  @SuppressWarnings("unchecked")
+  static ImmutableSetMultimap<String, PendingDeposit> decodePendings(String encodedPending)
+      throws IOException, ClassNotFoundException {
+    try (ObjectInputStream ois =
+        new ObjectInputStream(
+            new ByteArrayInputStream(
+                BaseEncoding.base64Url().omitPadding().decode(encodedPending)))) {
+      return (ImmutableSetMultimap<String, PendingDeposit>) ois.readObject();
+    }
+  }
+
+  /**
+   * Encodes the TLD to pending deposit map in an URL safe string that is sent to the pipeline
+   * worker by the pipeline launcher as a pipeline option.
+   */
+  public static String encodePendings(ImmutableSetMultimap<String, PendingDeposit> pendings)
+      throws IOException {
+    try (ByteArrayOutputStream baos = new ByteArrayOutputStream()) {
+      ObjectOutputStream oos = new ObjectOutputStream(baos);
+      oos.writeObject(pendings);
+      oos.flush();
+      return BaseEncoding.base64Url().omitPadding().encode(baos.toByteArray());
+    }
+  }
+
+  public static void main(String[] args) throws IOException, ClassNotFoundException {
+    PipelineOptionsFactory.register(RdePipelineOptions.class);
+    RdePipelineOptions options = PipelineOptionsFactory.fromArgs(args).as(RdePipelineOptions.class);
+    new RdePipeline(options).run();
+  }
+}
@@ -0,0 +1,32 @@
+// 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.beam.rde;
+
+import google.registry.beam.common.RegistryPipelineOptions;
+import org.apache.beam.sdk.options.Description;
+
+/** Custom options for running the spec11 pipeline. */
+public interface RdePipelineOptions extends RegistryPipelineOptions {
+
+  @Description("The Base64-encoded serialized map of TLDs to PendingDeposit.")
+  String getPendings();
+
+  void setPendings(String value);
+
+  @Description("The validation mode (LENIENT|STRICT) that the RDE marshaller uses.")
+  String getValidationMode();
+
+  void setValidationMode(String value);
+}
@@ -230,7 +230,7 @@
  </cron>

  <cron>
-    <url><![CDATA[/_dr/cron/fanout?queue=replay-commit-logs-to-sql&endpoint=/_dr/task/replayCommitLogsToSql&runInEmpty]]></url>
+    <url><![CDATA[/_dr/cron/fanout?queue=replay-commit-logs-to-sql&endpoint=/_dr/task/replayCommitLogsToSql&runInEmpty&dryRun=true]]></url>
    <description>
      Replays recent commit logs from Datastore to the SQL secondary backend.
    </description>
@@ -15,6 +15,7 @@
 package google.registry.flows;

 import static com.google.common.base.Preconditions.checkState;
+import static google.registry.persistence.transaction.TransactionManagerFactory.tm;

 import com.google.common.base.Strings;
 import dagger.Module;
@@ -222,6 +223,7 @@ public class FlowModule {
          String clientId,
          EppInput eppInput) {
    builder
+        .setModificationTime(tm().getTransactionTime())
        .setTrid(trid)
        .setXmlBytes(inputXmlBytes)
        .setBySuperuser(isSuperuser)
@@ -93,7 +93,6 @@ public final class ContactCreateFlow implements TransactionalFlow {
    validateContactAgainstPolicy(newContact);
    historyBuilder
        .setType(HistoryEntry.Type.CONTACT_CREATE)
-        .setModificationTime(now)
        .setXmlBytes(null) // We don't want to store contact details in the history entry.
        .setContact(newContact);
    tm().insertAll(
@@ -122,11 +122,7 @@ public final class ContactDeleteFlow implements TransactionalFlow {
      resultCode = SUCCESS;
    }
    ContactHistory contactHistory =
-        historyBuilder
-            .setType(historyEntryType)
-            .setModificationTime(now)
-            .setContact(newContact)
-            .build();
+        historyBuilder.setType(historyEntryType).setContact(newContact).build();
    if (!tm().isOfy()) {
      handlePendingTransferOnDelete(existingContact, newContact, now, contactHistory);
    }
@@ -22,6 +22,7 @@ import static google.registry.flows.ResourceFlowUtils.verifyResourceOwnership;
 import static google.registry.flows.contact.ContactFlowUtils.createGainingTransferPollMessage;
 import static google.registry.flows.contact.ContactFlowUtils.createTransferResponse;
 import static google.registry.model.ResourceTransferUtils.approvePendingTransfer;
+import static google.registry.model.reporting.HistoryEntry.Type.CONTACT_TRANSFER_APPROVE;
 import static google.registry.persistence.transaction.TransactionManagerFactory.tm;

 import com.google.common.collect.ImmutableSet;
@@ -39,7 +40,6 @@ import google.registry.model.eppcommon.AuthInfo;
 import google.registry.model.eppinput.ResourceCommand;
 import google.registry.model.eppoutput.EppResponse;
 import google.registry.model.poll.PollMessage;
-import google.registry.model.reporting.HistoryEntry;
 import google.registry.model.reporting.IcannReportingTypes.ActivityReportField;
 import google.registry.model.transfer.TransferStatus;
 import java.util.Optional;
@@ -88,11 +88,7 @@ public final class ContactTransferApproveFlow implements TransactionalFlow {
    ContactResource newContact =
        approvePendingTransfer(existingContact, TransferStatus.CLIENT_APPROVED, now);
    ContactHistory contactHistory =
-        historyBuilder
-            .setType(HistoryEntry.Type.CONTACT_TRANSFER_APPROVE)
-            .setModificationTime(now)
-            .setContact(newContact)
-            .build();
+        historyBuilder.setType(CONTACT_TRANSFER_APPROVE).setContact(newContact).build();
    // Create a poll message for the gaining client.
    PollMessage gainingPollMessage =
        createGainingTransferPollMessage(
@@ -22,6 +22,7 @@ import static google.registry.flows.ResourceFlowUtils.verifyTransferInitiator;
 import static google.registry.flows.contact.ContactFlowUtils.createLosingTransferPollMessage;
 import static google.registry.flows.contact.ContactFlowUtils.createTransferResponse;
 import static google.registry.model.ResourceTransferUtils.denyPendingTransfer;
+import static google.registry.model.reporting.HistoryEntry.Type.CONTACT_TRANSFER_CANCEL;
 import static google.registry.persistence.transaction.TransactionManagerFactory.tm;

 import com.google.common.collect.ImmutableSet;
@@ -39,7 +40,6 @@ import google.registry.model.eppcommon.AuthInfo;
 import google.registry.model.eppinput.ResourceCommand;
 import google.registry.model.eppoutput.EppResponse;
 import google.registry.model.poll.PollMessage;
-import google.registry.model.reporting.HistoryEntry;
 import google.registry.model.reporting.IcannReportingTypes.ActivityReportField;
 import google.registry.model.transfer.TransferStatus;
 import java.util.Optional;
@@ -84,11 +84,7 @@ public final class ContactTransferCancelFlow implements TransactionalFlow {
    ContactResource newContact =
        denyPendingTransfer(existingContact, TransferStatus.CLIENT_CANCELLED, now, clientId);
    ContactHistory contactHistory =
-        historyBuilder
-            .setType(HistoryEntry.Type.CONTACT_TRANSFER_CANCEL)
-            .setModificationTime(now)
-            .setContact(newContact)
-            .build();
+        historyBuilder.setType(CONTACT_TRANSFER_CANCEL).setContact(newContact).build();
    // Create a poll message for the losing client.
    PollMessage losingPollMessage =
        createLosingTransferPollMessage(
@@ -22,6 +22,7 @@ import static google.registry.flows.ResourceFlowUtils.verifyResourceOwnership;
 import static google.registry.flows.contact.ContactFlowUtils.createGainingTransferPollMessage;
 import static google.registry.flows.contact.ContactFlowUtils.createTransferResponse;
 import static google.registry.model.ResourceTransferUtils.denyPendingTransfer;
+import static google.registry.model.reporting.HistoryEntry.Type.CONTACT_TRANSFER_REJECT;
 import static google.registry.persistence.transaction.TransactionManagerFactory.tm;

 import com.google.common.collect.ImmutableSet;
@@ -38,7 +39,6 @@ import google.registry.model.domain.metadata.MetadataExtension;
 import google.registry.model.eppcommon.AuthInfo;
 import google.registry.model.eppoutput.EppResponse;
 import google.registry.model.poll.PollMessage;
-import google.registry.model.reporting.HistoryEntry;
 import google.registry.model.reporting.IcannReportingTypes.ActivityReportField;
 import google.registry.model.transfer.TransferStatus;
 import java.util.Optional;
@@ -82,11 +82,7 @@ public final class ContactTransferRejectFlow implements TransactionalFlow {
    ContactResource newContact =
        denyPendingTransfer(existingContact, TransferStatus.CLIENT_REJECTED, now, clientId);
    ContactHistory contactHistory =
-        historyBuilder
-            .setType(HistoryEntry.Type.CONTACT_TRANSFER_REJECT)
-            .setModificationTime(now)
-            .setContact(newContact)
-            .build();
+        historyBuilder.setType(CONTACT_TRANSFER_REJECT).setContact(newContact).build();
    PollMessage gainingPollMessage =
        createGainingTransferPollMessage(
            targetId, newContact.getTransferData(), now, Key.create(contactHistory));
@@ -24,6 +24,7 @@ import static google.registry.flows.contact.ContactFlowUtils.createGainingTransf
 import static google.registry.flows.contact.ContactFlowUtils.createLosingTransferPollMessage;
 import static google.registry.flows.contact.ContactFlowUtils.createTransferResponse;
 import static google.registry.model.eppoutput.Result.Code.SUCCESS_WITH_ACTION_PENDING;
+import static google.registry.model.reporting.HistoryEntry.Type.CONTACT_TRANSFER_REQUEST;
 import static google.registry.persistence.transaction.TransactionManagerFactory.tm;

 import com.google.common.collect.ImmutableSet;
@@ -45,7 +46,6 @@ import google.registry.model.eppcommon.StatusValue;
 import google.registry.model.eppcommon.Trid;
 import google.registry.model.eppoutput.EppResponse;
 import google.registry.model.poll.PollMessage;
-import google.registry.model.reporting.HistoryEntry;
 import google.registry.model.reporting.IcannReportingTypes.ActivityReportField;
 import google.registry.model.transfer.ContactTransferData;
 import google.registry.model.transfer.TransferStatus;
@@ -120,10 +120,7 @@ public final class ContactTransferRequestFlow implements TransactionalFlow {
            .setTransferStatus(TransferStatus.SERVER_APPROVED)
            .build();
    Key<ContactHistory> contactHistoryKey = createHistoryKey(existingContact, ContactHistory.class);
-    historyBuilder
-        .setId(contactHistoryKey.getId())
-        .setType(HistoryEntry.Type.CONTACT_TRANSFER_REQUEST)
-        .setModificationTime(now);
+    historyBuilder.setId(contactHistoryKey.getId()).setType(CONTACT_TRANSFER_REQUEST);
    // If the transfer is server approved, this message will be sent to the losing registrar. */
    PollMessage serverApproveLosingPollMessage =
        createLosingTransferPollMessage(targetId, serverApproveTransferData, contactHistoryKey);
@@ -24,6 +24,7 @@ import static google.registry.flows.ResourceFlowUtils.verifyOptionalAuthInfo;
 import static google.registry.flows.ResourceFlowUtils.verifyResourceOwnership;
 import static google.registry.flows.contact.ContactFlowUtils.validateAsciiPostalInfo;
 import static google.registry.flows.contact.ContactFlowUtils.validateContactAgainstPolicy;
+import static google.registry.model.reporting.HistoryEntry.Type.CONTACT_UPDATE;
 import static google.registry.persistence.transaction.TransactionManagerFactory.tm;

 import com.google.common.collect.ImmutableSet;
@@ -45,7 +46,6 @@ import google.registry.model.eppcommon.AuthInfo;
 import google.registry.model.eppcommon.StatusValue;
 import google.registry.model.eppinput.ResourceCommand;
 import google.registry.model.eppoutput.EppResponse;
-import google.registry.model.reporting.HistoryEntry;
 import google.registry.model.reporting.IcannReportingTypes.ActivityReportField;
 import java.util.Optional;
 import javax.annotation.Nullable;
@@ -146,8 +146,7 @@ public final class ContactUpdateFlow implements TransactionalFlow {
    validateAsciiPostalInfo(newContact.getInternationalizedPostalInfo());
    validateContactAgainstPolicy(newContact);
    historyBuilder
-        .setType(HistoryEntry.Type.CONTACT_UPDATE)
-        .setModificationTime(now)
+        .setType(CONTACT_UPDATE)
        .setXmlBytes(null) // We don't want to store contact details in the history entry.
        .setContact(newContact);
    tm().insert(historyBuilder.build());
@@ -48,6 +48,7 @@ import static google.registry.model.registry.Registry.TldState.GENERAL_AVAILABIL
 import static google.registry.model.registry.Registry.TldState.QUIET_PERIOD;
 import static google.registry.model.registry.Registry.TldState.START_DATE_SUNRISE;
 import static google.registry.model.registry.label.ReservationType.NAME_COLLISION;
+import static google.registry.model.reporting.HistoryEntry.Type.DOMAIN_CREATE;
 import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
 import static google.registry.util.DateTimeUtils.END_OF_TIME;
 import static google.registry.util.DateTimeUtils.leapSafeAddYears;
@@ -495,12 +496,7 @@ public class DomainCreateFlow implements TransactionalFlow {
                      TransactionReportField.netAddsFieldFromYears(period.getValue()),
                      1)));
    }
-    return historyBuilder
-        .setType(HistoryEntry.Type.DOMAIN_CREATE)
-        .setPeriod(period)
-        .setModificationTime(now)
-        .setDomain(domain)
-        .build();
+    return historyBuilder.setType(DOMAIN_CREATE).setPeriod(period).setDomain(domain).build();
  }

  private BillingEvent.OneTime createOneTimeBillingEvent(
@@ -34,6 +34,7 @@ import static google.registry.model.eppoutput.Result.Code.SUCCESS;
 import static google.registry.model.eppoutput.Result.Code.SUCCESS_WITH_ACTION_PENDING;
 import static google.registry.model.reporting.DomainTransactionRecord.TransactionReportField.ADD_FIELDS;
 import static google.registry.model.reporting.DomainTransactionRecord.TransactionReportField.RENEW_FIELDS;
+import static google.registry.model.reporting.HistoryEntry.Type.DOMAIN_DELETE;
 import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
 import static google.registry.pricing.PricingEngineProxy.getDomainRenewCost;
 import static google.registry.util.CollectionUtils.nullToEmpty;
@@ -88,7 +89,6 @@ import google.registry.model.registry.Registry;
 import google.registry.model.registry.Registry.TldType;
 import google.registry.model.reporting.DomainTransactionRecord;
 import google.registry.model.reporting.DomainTransactionRecord.TransactionReportField;
-import google.registry.model.reporting.HistoryEntry;
 import google.registry.model.reporting.IcannReportingTypes.ActivityReportField;
 import google.registry.model.transfer.TransferStatus;
 import java.util.Collections;
@@ -331,11 +331,7 @@ public final class DomainDeleteFlow implements TransactionalFlow {
                      : TransactionReportField.DELETED_DOMAINS_NOGRACE,
                  1)));
    }
-    return historyBuilder
-        .setType(HistoryEntry.Type.DOMAIN_DELETE)
-        .setModificationTime(now)
-        .setDomain(domain)
-        .build();
+    return historyBuilder.setType(DOMAIN_DELETE).setDomain(domain).build();
  }

  private PollMessage.OneTime createDeletePollMessage(
@@ -29,6 +29,7 @@ import static google.registry.flows.domain.DomainFlowUtils.validateFeeChallenge;
 import static google.registry.flows.domain.DomainFlowUtils.validateRegistrationPeriod;
 import static google.registry.flows.domain.DomainFlowUtils.verifyRegistrarIsActive;
 import static google.registry.flows.domain.DomainFlowUtils.verifyUnitIsYears;
+import static google.registry.model.reporting.HistoryEntry.Type.DOMAIN_RENEW;
 import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
 import static google.registry.util.DateTimeUtils.leapSafeAddYears;

@@ -73,7 +74,6 @@ import google.registry.model.poll.PollMessage;
 import google.registry.model.registry.Registry;
 import google.registry.model.reporting.DomainTransactionRecord;
 import google.registry.model.reporting.DomainTransactionRecord.TransactionReportField;
-import google.registry.model.reporting.HistoryEntry;
 import google.registry.model.reporting.IcannReportingTypes.ActivityReportField;
 import java.util.Optional;
 import javax.inject.Inject;
@@ -230,9 +230,8 @@ public final class DomainRenewFlow implements TransactionalFlow {
  private DomainHistory buildDomainHistory(
      DomainBase newDomain, DateTime now, Period period, Duration renewGracePeriod) {
    return historyBuilder
-        .setType(HistoryEntry.Type.DOMAIN_RENEW)
+        .setType(DOMAIN_RENEW)
        .setPeriod(period)
-        .setModificationTime(now)
        .setDomain(newDomain)
        .setDomainTransactionRecords(
            ImmutableSet.of(
@@ -27,6 +27,7 @@ import static google.registry.flows.domain.DomainFlowUtils.verifyNotReserved;
 import static google.registry.flows.domain.DomainFlowUtils.verifyPremiumNameIsNotBlocked;
 import static google.registry.flows.domain.DomainFlowUtils.verifyRegistrarIsActive;
 import static google.registry.model.ResourceTransferUtils.updateForeignKeyIndexDeletionTime;
+import static google.registry.model.reporting.HistoryEntry.Type.DOMAIN_RESTORE;
 import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
 import static google.registry.util.DateTimeUtils.END_OF_TIME;

@@ -66,7 +67,6 @@ import google.registry.model.poll.PollMessage;
 import google.registry.model.registry.Registry;
 import google.registry.model.reporting.DomainTransactionRecord;
 import google.registry.model.reporting.DomainTransactionRecord.TransactionReportField;
-import google.registry.model.reporting.HistoryEntry;
 import google.registry.model.reporting.IcannReportingTypes.ActivityReportField;
 import java.util.Optional;
 import javax.inject.Inject;
@@ -188,8 +188,7 @@ public final class DomainRestoreRequestFlow implements TransactionalFlow  {

  private DomainHistory buildDomainHistory(DomainBase newDomain, DateTime now) {
    return historyBuilder
-        .setType(HistoryEntry.Type.DOMAIN_RESTORE)
-        .setModificationTime(now)
+        .setType(DOMAIN_RESTORE)
        .setDomain(newDomain)
        .setDomainTransactionRecords(
            ImmutableSet.of(
@@ -29,6 +29,7 @@ import static google.registry.flows.domain.DomainTransferUtils.createGainingTran
 import static google.registry.flows.domain.DomainTransferUtils.createTransferResponse;
 import static google.registry.model.ResourceTransferUtils.approvePendingTransfer;
 import static google.registry.model.reporting.DomainTransactionRecord.TransactionReportField.TRANSFER_SUCCESSFUL;
+import static google.registry.model.reporting.HistoryEntry.Type.DOMAIN_TRANSFER_APPROVE;
 import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
 import static google.registry.pricing.PricingEngineProxy.getDomainRenewCost;
 import static google.registry.util.CollectionUtils.union;
@@ -58,7 +59,6 @@ import google.registry.model.eppoutput.EppResponse;
 import google.registry.model.poll.PollMessage;
 import google.registry.model.registry.Registry;
 import google.registry.model.reporting.DomainTransactionRecord;
-import google.registry.model.reporting.HistoryEntry;
 import google.registry.model.reporting.IcannReportingTypes.ActivityReportField;
 import google.registry.model.transfer.DomainTransferData;
 import google.registry.model.transfer.TransferStatus;
@@ -240,8 +240,7 @@ public final class DomainTransferApproveFlow implements TransactionalFlow {
            registry.getAutomaticTransferLength().plus(registry.getTransferGracePeriodLength()),
            ImmutableSet.of(TRANSFER_SUCCESSFUL));
    return historyBuilder
-        .setType(HistoryEntry.Type.DOMAIN_TRANSFER_APPROVE)
-        .setModificationTime(now)
+        .setType(DOMAIN_TRANSFER_APPROVE)
        .setOtherClientId(gainingClientId)
        .setDomain(newDomain)
        .setDomainTransactionRecords(
@@ -27,6 +27,7 @@ import static google.registry.flows.domain.DomainTransferUtils.createLosingTrans
 import static google.registry.flows.domain.DomainTransferUtils.createTransferResponse;
 import static google.registry.model.ResourceTransferUtils.denyPendingTransfer;
 import static google.registry.model.reporting.DomainTransactionRecord.TransactionReportField.TRANSFER_SUCCESSFUL;
+import static google.registry.model.reporting.HistoryEntry.Type.DOMAIN_TRANSFER_CANCEL;
 import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
 import static google.registry.util.DateTimeUtils.END_OF_TIME;

@@ -46,7 +47,6 @@ import google.registry.model.eppcommon.AuthInfo;
 import google.registry.model.eppoutput.EppResponse;
 import google.registry.model.registry.Registry;
 import google.registry.model.reporting.DomainTransactionRecord;
-import google.registry.model.reporting.HistoryEntry;
 import google.registry.model.reporting.IcannReportingTypes.ActivityReportField;
 import google.registry.model.transfer.TransferStatus;
 import java.util.Optional;
@@ -130,8 +130,7 @@ public final class DomainTransferCancelFlow implements TransactionalFlow {
            registry.getAutomaticTransferLength().plus(registry.getTransferGracePeriodLength()),
            ImmutableSet.of(TRANSFER_SUCCESSFUL));
    return historyBuilder
-        .setType(HistoryEntry.Type.DOMAIN_TRANSFER_CANCEL)
-        .setModificationTime(now)
+        .setType(DOMAIN_TRANSFER_CANCEL)
        .setDomain(newDomain)
        .setDomainTransactionRecords(cancelingRecords)
        .build();
@@ -28,6 +28,7 @@ import static google.registry.flows.domain.DomainTransferUtils.createTransferRes
 import static google.registry.model.ResourceTransferUtils.denyPendingTransfer;
 import static google.registry.model.reporting.DomainTransactionRecord.TransactionReportField.TRANSFER_NACKED;
 import static google.registry.model.reporting.DomainTransactionRecord.TransactionReportField.TRANSFER_SUCCESSFUL;
+import static google.registry.model.reporting.HistoryEntry.Type.DOMAIN_TRANSFER_REJECT;
 import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
 import static google.registry.util.CollectionUtils.union;
 import static google.registry.util.DateTimeUtils.END_OF_TIME;
@@ -48,7 +49,6 @@ import google.registry.model.eppcommon.AuthInfo;
 import google.registry.model.eppoutput.EppResponse;
 import google.registry.model.registry.Registry;
 import google.registry.model.reporting.DomainTransactionRecord;
-import google.registry.model.reporting.HistoryEntry;
 import google.registry.model.reporting.IcannReportingTypes.ActivityReportField;
 import google.registry.model.transfer.TransferStatus;
 import java.util.Optional;
@@ -131,8 +131,7 @@ public final class DomainTransferRejectFlow implements TransactionalFlow {
            registry.getAutomaticTransferLength().plus(registry.getTransferGracePeriodLength()),
            ImmutableSet.of(TRANSFER_SUCCESSFUL));
    return historyBuilder
-        .setType(HistoryEntry.Type.DOMAIN_TRANSFER_REJECT)
-        .setModificationTime(now)
+        .setType(DOMAIN_TRANSFER_REJECT)
        .setDomainTransactionRecords(
            union(
                cancelingRecords,
@@ -32,6 +32,7 @@ import static google.registry.flows.domain.DomainTransferUtils.createPendingTran
 import static google.registry.flows.domain.DomainTransferUtils.createTransferResponse;
 import static google.registry.flows.domain.DomainTransferUtils.createTransferServerApproveEntities;
 import static google.registry.model.eppoutput.Result.Code.SUCCESS_WITH_ACTION_PENDING;
+import static google.registry.model.reporting.HistoryEntry.Type.DOMAIN_TRANSFER_REQUEST;
 import static google.registry.persistence.transaction.TransactionManagerFactory.tm;

 import com.google.common.collect.ImmutableList;
@@ -68,7 +69,6 @@ import google.registry.model.poll.PollMessage;
 import google.registry.model.registry.Registry;
 import google.registry.model.reporting.DomainTransactionRecord;
 import google.registry.model.reporting.DomainTransactionRecord.TransactionReportField;
-import google.registry.model.reporting.HistoryEntry;
 import google.registry.model.reporting.IcannReportingTypes.ActivityReportField;
 import google.registry.model.transfer.DomainTransferData;
 import google.registry.model.transfer.TransferData.TransferServerApproveEntity;
@@ -315,9 +315,8 @@ public final class DomainTransferRequestFlow implements TransactionalFlow {
  private DomainHistory buildDomainHistory(
      DomainBase newDomain, Registry registry, DateTime now, Period period) {
    return historyBuilder
-        .setType(HistoryEntry.Type.DOMAIN_TRANSFER_REQUEST)
+        .setType(DOMAIN_TRANSFER_REQUEST)
        .setPeriod(period)
-        .setModificationTime(now)
        .setDomain(newDomain)
        .setDomainTransactionRecords(
            ImmutableSet.of(
@@ -39,6 +39,7 @@ import static google.registry.flows.domain.DomainFlowUtils.validateRegistrantAll
 import static google.registry.flows.domain.DomainFlowUtils.validateRequiredContactsPresent;
 import static google.registry.flows.domain.DomainFlowUtils.verifyClientUpdateNotProhibited;
 import static google.registry.flows.domain.DomainFlowUtils.verifyNotInPendingDelete;
+import static google.registry.model.reporting.HistoryEntry.Type.DOMAIN_UPDATE;
 import static google.registry.persistence.transaction.TransactionManagerFactory.tm;

 import com.google.common.collect.ImmutableSet;
@@ -76,7 +77,6 @@ import google.registry.model.eppinput.EppInput;
 import google.registry.model.eppinput.ResourceCommand;
 import google.registry.model.eppoutput.EppResponse;
 import google.registry.model.registry.Registry;
-import google.registry.model.reporting.HistoryEntry;
 import google.registry.model.reporting.IcannReportingTypes.ActivityReportField;
 import java.util.Optional;
 import javax.inject.Inject;
@@ -166,7 +166,8 @@ public final class DomainUpdateFlow implements TransactionalFlow {
    flowCustomLogic.afterValidation(
        AfterValidationParameters.newBuilder().setExistingDomain(existingDomain).build());
    DomainBase newDomain = performUpdate(command, existingDomain, now);
-    DomainHistory domainHistory = buildDomainHistory(newDomain, now);
+    DomainHistory domainHistory =
+        historyBuilder.setType(DOMAIN_UPDATE).setDomain(newDomain).build();
    validateNewState(newDomain);
    dnsQueue.addDomainRefreshTask(targetId);
    ImmutableSet.Builder<ImmutableObject> entitiesToSave = new ImmutableSet.Builder<>();
@@ -217,14 +218,6 @@ public final class DomainUpdateFlow implements TransactionalFlow {
        tld, add.getNameserverFullyQualifiedHostNames());
  }

-  private DomainHistory buildDomainHistory(DomainBase newDomain, DateTime now) {
-    return historyBuilder
-        .setType(HistoryEntry.Type.DOMAIN_UPDATE)
-        .setModificationTime(now)
-        .setDomain(newDomain)
-        .build();
-  }
-
  private DomainBase performUpdate(Update command, DomainBase domain, DateTime now)
      throws EppException {
    AddRemove add = command.getInnerAdd();
@@ -22,6 +22,7 @@ import static google.registry.flows.host.HostFlowUtils.verifySuperordinateDomain
 import static google.registry.flows.host.HostFlowUtils.verifySuperordinateDomainOwnership;
 import static google.registry.model.EppResourceUtils.createRepoId;
 import static google.registry.model.IdService.allocateId;
+import static google.registry.model.reporting.HistoryEntry.Type.HOST_CREATE;
 import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
 import static google.registry.util.CollectionUtils.isNullOrEmpty;

@@ -50,7 +51,6 @@ import google.registry.model.host.HostHistory;
 import google.registry.model.host.HostResource;
 import google.registry.model.index.EppResourceIndex;
 import google.registry.model.index.ForeignKeyIndex;
-import google.registry.model.reporting.HistoryEntry;
 import google.registry.model.reporting.IcannReportingTypes.ActivityReportField;
 import java.util.Optional;
 import javax.inject.Inject;
@@ -129,7 +129,7 @@ public final class HostCreateFlow implements TransactionalFlow {
            .setRepoId(createRepoId(allocateId(), roidSuffix))
            .setSuperordinateDomain(superordinateDomain.map(DomainBase::createVKey).orElse(null))
            .build();
-    historyBuilder.setType(HistoryEntry.Type.HOST_CREATE).setModificationTime(now).setHost(newHost);
+    historyBuilder.setType(HOST_CREATE).setHost(newHost);
    ImmutableSet<ImmutableObject> entitiesToSave =
        ImmutableSet.of(
            newHost,
@@ -130,7 +130,7 @@ public final class HostDeleteFlow implements TransactionalFlow {
      historyEntryType = Type.HOST_DELETE;
      resultCode = SUCCESS;
    }
-    historyBuilder.setType(historyEntryType).setModificationTime(now).setHost(newHost);
+    historyBuilder.setType(historyEntryType).setHost(newHost);
    tm().insert(historyBuilder.build());
    tm().update(newHost);
    return responseBuilder.setResultFromCode(resultCode).build();
@@ -27,6 +27,7 @@ import static google.registry.flows.host.HostFlowUtils.validateHostName;
 import static google.registry.flows.host.HostFlowUtils.verifySuperordinateDomainNotInPendingDelete;
 import static google.registry.flows.host.HostFlowUtils.verifySuperordinateDomainOwnership;
 import static google.registry.model.index.ForeignKeyIndex.loadAndGetKey;
+import static google.registry.model.reporting.HistoryEntry.Type.HOST_UPDATE;
 import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
 import static google.registry.util.CollectionUtils.isNullOrEmpty;

@@ -57,7 +58,6 @@ import google.registry.model.host.HostCommand.Update.Change;
 import google.registry.model.host.HostHistory;
 import google.registry.model.host.HostResource;
 import google.registry.model.index.ForeignKeyIndex;
-import google.registry.model.reporting.HistoryEntry;
 import google.registry.model.reporting.IcannReportingTypes.ActivityReportField;
 import google.registry.persistence.VKey;
 import java.util.Objects;
@@ -201,12 +201,7 @@ public final class HostUpdateFlow implements TransactionalFlow {
      updateSuperordinateDomains(existingHost, newHost);
    }
    enqueueTasks(existingHost, newHost);
-    entitiesToInsert.add(
-        historyBuilder
-            .setType(HistoryEntry.Type.HOST_UPDATE)
-            .setModificationTime(now)
-            .setHost(newHost)
-            .build());
+    entitiesToInsert.add(historyBuilder.setType(HOST_UPDATE).setHost(newHost).build());
    tm().updateAll(entitiesToUpdate.build());
    tm().insertAll(entitiesToInsert.build());
    return responseBuilder.build();
@@ -398,7 +398,7 @@ public final class EppResourceUtils {
                .orElse(null);
    if (resourceAtPointInTime == null) {
      logger.atSevere().log(
-          "Couldn't load resource at % for key %s, falling back to resource %s.",
+          "Couldn't load resource at %s for key %s, falling back to resource %s.",
          timestamp, resource.createVKey(), resource);
      return resource;
    }
@@ -110,7 +110,8 @@ public class ContactHistory extends HistoryEntry implements SqlEntity {

  @Override
  public Optional<? extends EppResource> getResourceAtPointInTime() {
-    return getContactBase();
+    return getContactBase()
+        .map(contactBase -> new ContactResource.Builder().copyFrom(contactBase).build());
  }

  @PostLoad
@@ -79,5 +79,26 @@ public class ContactResource extends ContactBase
    private Builder(ContactResource instance) {
      super(instance);
    }
+
+    public Builder copyFrom(ContactBase contactBase) {
+      return this.setAuthInfo(contactBase.getAuthInfo())
+          .setContactId(contactBase.getContactId())
+          .setCreationClientId(contactBase.getCreationClientId())
+          .setCreationTime(contactBase.getCreationTime())
+          .setDeletionTime(contactBase.getDeletionTime())
+          .setDisclose(contactBase.getDisclose())
+          .setEmailAddress(contactBase.getEmailAddress())
+          .setFaxNumber(contactBase.getFaxNumber())
+          .setInternationalizedPostalInfo(contactBase.getInternationalizedPostalInfo())
+          .setLastTransferTime(contactBase.getLastTransferTime())
+          .setLastEppUpdateClientId(contactBase.getLastEppUpdateClientId())
+          .setLastEppUpdateTime(contactBase.getLastEppUpdateTime())
+          .setLocalizedPostalInfo(contactBase.getLocalizedPostalInfo())
+          .setPersistedCurrentSponsorClientId(contactBase.getPersistedCurrentSponsorClientId())
+          .setRepoId(contactBase.getRepoId())
+          .setStatusValues(contactBase.getStatusValues())
+          .setTransferData(contactBase.getTransferData())
+          .setVoiceNumber(contactBase.getVoiceNumber());
+    }
  }
 }
@@ -181,5 +181,34 @@ public class DomainBase extends DomainContent
    Builder(DomainBase instance) {
      super(instance);
    }
+
+    public Builder copyFrom(DomainContent domainContent) {
+      return this.setAuthInfo(domainContent.getAuthInfo())
+          .setAutorenewPollMessage(domainContent.getAutorenewPollMessage())
+          .setAutorenewBillingEvent(domainContent.getAutorenewBillingEvent())
+          .setAutorenewEndTime(domainContent.getAutorenewEndTime())
+          .setContacts(domainContent.getContacts())
+          .setCreationClientId(domainContent.getCreationClientId())
+          .setCreationTime(domainContent.getCreationTime())
+          .setDomainName(domainContent.getDomainName())
+          .setDeletePollMessage(domainContent.getDeletePollMessage())
+          .setDsData(domainContent.getDsData())
+          .setDeletionTime(domainContent.getDeletionTime())
+          .setGracePeriods(domainContent.getGracePeriods())
+          .setIdnTableName(domainContent.getIdnTableName())
+          .setLastTransferTime(domainContent.getLastTransferTime())
+          .setLaunchNotice(domainContent.getLaunchNotice())
+          .setLastEppUpdateClientId(domainContent.getLastEppUpdateClientId())
+          .setLastEppUpdateTime(domainContent.getLastEppUpdateTime())
+          .setNameservers(domainContent.getNameservers())
+          .setPersistedCurrentSponsorClientId(domainContent.getPersistedCurrentSponsorClientId())
+          .setRegistrant(domainContent.getRegistrant())
+          .setRegistrationExpirationTime(domainContent.getRegistrationExpirationTime())
+          .setRepoId(domainContent.getRepoId())
+          .setSmdId(domainContent.getSmdId())
+          .setSubordinateHosts(domainContent.getSubordinateHosts())
+          .setStatusValues(domainContent.getStatusValues())
+          .setTransferData(domainContent.getTransferData());
+    }
  }
 }
@@ -252,11 +252,18 @@ public class DomainHistory extends HistoryEntry implements SqlEntity {

  @Override
  public Optional<? extends EppResource> getResourceAtPointInTime() {
-    return getDomainContent();
+    return getDomainContent()
+        .map(domainContent -> new DomainBase.Builder().copyFrom(domainContent).build());
  }

  @PostLoad
  void postLoad() {
+    // TODO(b/188044616): Determine why Eager loading doesn't work here.
+    Hibernate.initialize(domainTransactionRecords);
+    Hibernate.initialize(nsHosts);
+    Hibernate.initialize(dsDataHistories);
+    Hibernate.initialize(gracePeriodHistories);
+
    if (domainContent != null) {
      domainContent.nsHosts = nullToEmptyImmutableCopy(nsHosts);
      domainContent.gracePeriods =
@@ -278,12 +285,6 @@ public class DomainHistory extends HistoryEntry implements SqlEntity {
        }
      }
    }
-
-    // TODO(b/188044616): Determine why Eager loading doesn't work here.
-    Hibernate.initialize(domainTransactionRecords);
-    Hibernate.initialize(nsHosts);
-    Hibernate.initialize(dsDataHistories);
-    Hibernate.initialize(gracePeriodHistories);
  }

  // In Datastore, save as a HistoryEntry object regardless of this object's type
@@ -137,7 +137,7 @@ public class HostBase extends EppResource {
  }

  @Override
-  public Builder asBuilder() {
+  public Builder<? extends HostBase, ?> asBuilder() {
    return new Builder<>(clone(this));
  }

@@ -111,7 +111,7 @@ public class HostHistory extends HistoryEntry implements SqlEntity {

  @Override
  public Optional<? extends EppResource> getResourceAtPointInTime() {
-    return getHostBase();
+    return getHostBase().map(hostBase -> new HostResource.Builder().copyFrom(hostBase).build());
  }

  @PostLoad
@@ -63,5 +63,21 @@ public class HostResource extends HostBase
    private Builder(HostResource instance) {
      super(instance);
    }
+
+    public Builder copyFrom(HostBase hostBase) {
+      return this.setCreationClientId(hostBase.getCreationClientId())
+          .setCreationTime(hostBase.getCreationTime())
+          .setDeletionTime(hostBase.getDeletionTime())
+          .setHostName(hostBase.getHostName())
+          .setInetAddresses(hostBase.getInetAddresses())
+          .setLastTransferTime(hostBase.getLastTransferTime())
+          .setLastSuperordinateChange(hostBase.getLastSuperordinateChange())
+          .setLastEppUpdateClientId(hostBase.getLastEppUpdateClientId())
+          .setLastEppUpdateTime(hostBase.getLastEppUpdateTime())
+          .setPersistedCurrentSponsorClientId(hostBase.getPersistedCurrentSponsorClientId())
+          .setRepoId(hostBase.getRepoId())
+          .setSuperordinateDomain(hostBase.getSuperordinateDomain())
+          .setStatusValues(hostBase.getStatusValues());
+    }
  }
 }
@@ -17,13 +17,14 @@ package google.registry.model.translators;
 import static com.google.common.base.MoreObjects.firstNonNull;
 import static google.registry.config.RegistryConfig.getCommitLogDatastoreRetention;
 import static google.registry.model.ofy.ObjectifyService.auditedOfy;
-import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
+import static google.registry.persistence.transaction.TransactionManagerFactory.ofyTm;
 import static google.registry.util.DateTimeUtils.START_OF_TIME;

 import com.google.common.collect.ImmutableSortedMap;
 import com.google.common.collect.Ordering;
 import com.googlecode.objectify.Key;
 import google.registry.model.ofy.CommitLogManifest;
+import google.registry.persistence.transaction.Transaction;
 import org.joda.time.DateTime;

 /**
@@ -58,12 +59,20 @@ public final class CommitLogRevisionsTranslatorFactory
   * <p>We store a maximum of one entry per day. It will be the last transaction that happened on
   * that day.
   *
+   * <p>In serialization mode, this method just returns "revisions" without modification.
+   *
   * @see google.registry.config.RegistryConfig#getCommitLogDatastoreRetention()
   */
  @Override
  ImmutableSortedMap<DateTime, Key<CommitLogManifest>> transformBeforeSave(
      ImmutableSortedMap<DateTime, Key<CommitLogManifest>> revisions) {
-    DateTime now = tm().getTransactionTime();
+
+    // Don't do anything if we're just doing object serialization.
+    if (Transaction.inSerializationMode()) {
+      return revisions;
+    }
+
+    DateTime now = ofyTm().getTransactionTime();
    DateTime threshold = now.minus(getCommitLogDatastoreRetention());
    DateTime preThresholdTime = firstNonNull(revisions.floorKey(threshold), START_OF_TIME);
    return new ImmutableSortedMap.Builder<DateTime, Key<CommitLogManifest>>(Ordering.natural())
@@ -15,10 +15,11 @@
 package google.registry.model.translators;

 import static com.google.common.base.MoreObjects.firstNonNull;
-import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
+import static google.registry.persistence.transaction.TransactionManagerFactory.ofyTm;
 import static org.joda.time.DateTimeZone.UTC;

 import google.registry.model.CreateAutoTimestamp;
+import google.registry.persistence.transaction.Transaction;
 import java.util.Date;
 import org.joda.time.DateTime;

@@ -46,7 +47,14 @@ public class CreateAutoTimestampTranslatorFactory
      /** Save a timestamp, setting it to the current time if it did not have a previous value. */
      @Override
      public Date saveValue(CreateAutoTimestamp pojoValue) {
-        return firstNonNull(pojoValue.getTimestamp(), tm().getTransactionTime()).toDate();
-      }};
+
+        // Don't do this if we're in the course of transaction serialization.
+        if (Transaction.inSerializationMode()) {
+          return pojoValue.getTimestamp() == null ? null : pojoValue.getTimestamp().toDate();
+        }
+
+        return firstNonNull(pojoValue.getTimestamp(), ofyTm().getTransactionTime()).toDate();
+      }
+    };
  }
 }
@@ -14,10 +14,11 @@

 package google.registry.model.translators;

-import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
+import static google.registry.persistence.transaction.TransactionManagerFactory.ofyTm;
 import static org.joda.time.DateTimeZone.UTC;

 import google.registry.model.UpdateAutoTimestamp;
+import google.registry.persistence.transaction.Transaction;
 import java.util.Date;
 import org.joda.time.DateTime;

@@ -46,8 +47,14 @@ public class UpdateAutoTimestampTranslatorFactory
      /** Save a timestamp, setting it to the current time. */
      @Override
      public Date saveValue(UpdateAutoTimestamp pojoValue) {
+
+        // Don't do this if we're in the course of transaction serialization.
+        if (Transaction.inSerializationMode()) {
+          return pojoValue.getTimestamp() == null ? null : pojoValue.getTimestamp().toDate();
+        }
+
        return UpdateAutoTimestamp.autoUpdateEnabled()
-            ? tm().getTransactionTime().toDate()
+            ? ofyTm().getTransactionTime().toDate()
            : pojoValue.getTimestamp().toDate();
      }
    };
@@ -50,6 +50,11 @@ public interface JpaTransactionManager extends TransactionManager {
   */
  Query query(String sqlString);

+  /**
+   * Execute the work in a transaction without recording the transaction for replay to datastore.
+   */
+  <T> T transactWithoutBackup(Supplier<T> work);
+
  /** Executes the work in a transaction with no retries and returns the result. */
  <T> T transactNoRetry(Supplier<T> work);

@@ -41,6 +41,8 @@ import google.registry.model.server.KmsSecret;
 import google.registry.model.tmch.ClaimsList.ClaimsListSingleton;
 import google.registry.persistence.JpaRetries;
 import google.registry.persistence.VKey;
+import google.registry.schema.replay.NonReplicatedEntity;
+import google.registry.schema.replay.SqlOnlyEntity;
 import google.registry.util.Clock;
 import google.registry.util.Retrier;
 import google.registry.util.SystemSleeper;
@@ -152,6 +154,15 @@ public class JpaTransactionManagerImpl implements JpaTransactionManager {

  @Override
  public <T> T transact(Supplier<T> work) {
+    return transact(work, true);
+  }
+
+  @Override
+  public <T> T transactWithoutBackup(Supplier<T> work) {
+    return transact(work, false);
+  }
+
+  private <T> T transact(Supplier<T> work, boolean withBackup) {
    return retrier.callWithRetry(
        () -> {
          if (inTransaction()) {
@@ -162,7 +173,7 @@ public class JpaTransactionManagerImpl implements JpaTransactionManager {
          EntityTransaction txn = txnInfo.entityManager.getTransaction();
          try {
            txn.begin();
-            txnInfo.start(clock);
+            txnInfo.start(clock, withBackup);
            T result = work.get();
            txnInfo.recordTransaction();
            txn.commit();
@@ -194,7 +205,7 @@ public class JpaTransactionManagerImpl implements JpaTransactionManager {
    EntityTransaction txn = txnInfo.entityManager.getTransaction();
    try {
      txn.begin();
-      txnInfo.start(clock);
+      txnInfo.start(clock, true);
      T result = work.get();
      txnInfo.recordTransaction();
      txn.commit();
@@ -740,11 +751,11 @@ public class JpaTransactionManagerImpl implements JpaTransactionManager {
    Set<Object> objectsToSave = Collections.newSetFromMap(new IdentityHashMap<Object, Boolean>());

    /** Start a new transaction. */
-    private void start(Clock clock) {
+    private void start(Clock clock, boolean withBackup) {
      checkArgumentNotNull(clock);
      inTransaction = true;
      transactionTime = clock.nowUtc();
-      if (RegistryConfig.getCloudSqlReplicateTransactions()) {
+      if (withBackup && RegistryConfig.getCloudSqlReplicateTransactions()) {
        contentsBuilder = new Transaction.Builder();
      }
    }
@@ -763,17 +774,23 @@ public class JpaTransactionManagerImpl implements JpaTransactionManager {
    }

    private void addUpdate(Object entity) {
-      if (contentsBuilder != null) {
+      if (contentsBuilder != null && shouldReplicate(entity.getClass())) {
        contentsBuilder.addUpdate(entity);
      }
    }

    private void addDelete(VKey<?> key) {
-      if (contentsBuilder != null) {
+      if (contentsBuilder != null && shouldReplicate(key.getKind())) {
        contentsBuilder.addDelete(key);
      }
    }

+    /** Returns true if the entity class should be replicated from SQL to datastore. */
+    private boolean shouldReplicate(Class<?> entityClass) {
+      return !NonReplicatedEntity.class.isAssignableFrom(entityClass)
+          && !SqlOnlyEntity.class.isAssignableFrom(entityClass);
+    }
+
    private void recordTransaction() {
      if (contentsBuilder != null) {
        Transaction persistedTxn = contentsBuilder.build();
@@ -18,7 +18,6 @@ import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Preconditions.checkNotNull;
 import static google.registry.model.ofy.ObjectifyService.auditedOfy;
 import static google.registry.persistence.transaction.TransactionManagerFactory.ofyTm;
-import static google.registry.persistence.transaction.TransactionManagerFactory.tm;

 import com.google.appengine.api.datastore.Entity;
 import com.google.appengine.api.datastore.EntityTranslator;
@@ -51,11 +50,17 @@ public class Transaction extends ImmutableObject implements Buildable {
  // unique and inherently informative.
  private static final int VERSION_ID = 20200604;

+  // Keep a per-thread flag to keep track of whether we're serializing an entity for a transaction.
+  // This is used by internal translators to avoid doing things that are dependent on being in a
+  // datastore transaction and alter the persisted representation of the entity.
+  private static ThreadLocal<Boolean> inSerializationMode = ThreadLocal.withInitial(() -> false);
+
  private transient ImmutableList<Mutation> mutations;

  /** Write the entire transaction to the datastore in a datastore transaction. */
  public void writeToDatastore() {
-    tm().transact(
+    ofyTm()
+        .transact(
            () -> {
              for (Mutation mutation : mutations) {
                mutation.writeToDatastore();
@@ -75,8 +80,13 @@ public class Transaction extends ImmutableObject implements Buildable {

      // Write all of the mutations, preceded by their count.
      out.writeInt(mutations.size());
-      for (Mutation mutation : mutations) {
-        mutation.serializeTo(out);
+      try {
+        inSerializationMode.set(true);
+        for (Mutation mutation : mutations) {
+          mutation.serializeTo(out);
+        }
+      } finally {
+        inSerializationMode.set(false);
      }

      out.close();
@@ -114,6 +124,17 @@ public class Transaction extends ImmutableObject implements Buildable {
    return mutations.isEmpty();
  }

+  /**
+   * Returns true if we are serializing a transaction in the current thread.
+   *
+   * <p>This should be checked by any Ofy translators prior to making any changes to an entity's
+   * state representation based on the assumption that we are currently pseristing the entity to
+   * datastore.
+   */
+  public static boolean inSerializationMode() {
+    return inSerializationMode.get();
+  }
+
  @Override
  public Builder asBuilder() {
    return new Builder(clone(this));
@@ -17,8 +17,17 @@ package google.registry.rde;
 import com.google.auto.value.AutoValue;
 import google.registry.model.common.Cursor.CursorType;
 import google.registry.model.rde.RdeMode;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
 import java.io.Serializable;
 import javax.annotation.Nullable;
+import org.apache.beam.sdk.coders.AtomicCoder;
+import org.apache.beam.sdk.coders.BooleanCoder;
+import org.apache.beam.sdk.coders.NullableCoder;
+import org.apache.beam.sdk.coders.SerializableCoder;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.coders.VarIntCoder;
 import org.joda.time.DateTime;
 import org.joda.time.Duration;

@@ -69,17 +78,9 @@ public abstract class PendingDeposit implements Serializable {
  @Nullable
  public abstract Integer revision();

-  static PendingDeposit create(
+  public static PendingDeposit create(
      String tld, DateTime watermark, RdeMode mode, CursorType cursor, Duration interval) {
-    return new AutoValue_PendingDeposit(
-        false,
-        tld,
-        watermark,
-        mode,
-        cursor,
-        interval,
-        null,
-        null);
+    return new AutoValue_PendingDeposit(false, tld, watermark, mode, cursor, interval, null, null);
  }

  static PendingDeposit createInManualOperation(
@@ -89,15 +90,52 @@ public abstract class PendingDeposit implements Serializable {
      String directoryWithTrailingSlash,
      @Nullable Integer revision) {
    return new AutoValue_PendingDeposit(
-        true,
-        tld,
-        watermark,
-        mode,
-        null,
-        null,
-        directoryWithTrailingSlash,
-        revision);
+        true, tld, watermark, mode, null, null, directoryWithTrailingSlash, revision);
  }

  PendingDeposit() {}
+
+  /**
+   * A deterministic coder for {@link PendingDeposit} used during a GroupBy transform.
+   *
+   * <p>We cannot use a {@link SerializableCoder} directly because it does not guarantee
+   * determinism, which is required by GroupBy.
+   */
+  public static class PendingDepositCoder extends AtomicCoder<PendingDeposit> {
+
+    private PendingDepositCoder() {
+      super();
+    }
+
+    private static final PendingDepositCoder INSTANCE = new PendingDepositCoder();
+
+    public static PendingDepositCoder of() {
+      return INSTANCE;
+    }
+
+    @Override
+    public void encode(PendingDeposit value, OutputStream outStream) throws IOException {
+      BooleanCoder.of().encode(value.manual(), outStream);
+      StringUtf8Coder.of().encode(value.tld(), outStream);
+      SerializableCoder.of(DateTime.class).encode(value.watermark(), outStream);
+      SerializableCoder.of(RdeMode.class).encode(value.mode(), outStream);
+      NullableCoder.of(SerializableCoder.of(CursorType.class)).encode(value.cursor(), outStream);
+      NullableCoder.of(SerializableCoder.of(Duration.class)).encode(value.interval(), outStream);
+      NullableCoder.of(StringUtf8Coder.of()).encode(value.directoryWithTrailingSlash(), outStream);
+      NullableCoder.of(VarIntCoder.of()).encode(value.revision(), outStream);
+    }
+
+    @Override
+    public PendingDeposit decode(InputStream inStream) throws IOException {
+      return new AutoValue_PendingDeposit(
+          BooleanCoder.of().decode(inStream),
+          StringUtf8Coder.of().decode(inStream),
+          SerializableCoder.of(DateTime.class).decode(inStream),
+          SerializableCoder.of(RdeMode.class).decode(inStream),
+          NullableCoder.of(SerializableCoder.of(CursorType.class)).decode(inStream),
+          NullableCoder.of(SerializableCoder.of(Duration.class)).decode(inStream),
+          NullableCoder.of(StringUtf8Coder.of()).decode(inStream),
+          NullableCoder.of(VarIntCoder.of()).decode(inStream));
+    }
+  }
 }
@@ -0,0 +1,109 @@
+// 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.rde;
+
+import static google.registry.model.EppResourceUtils.loadAtPointInTime;
+import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
+
+import com.google.auto.value.AutoValue;
+import com.google.common.collect.ImmutableMap;
+import google.registry.model.EppResource;
+import google.registry.model.contact.ContactResource;
+import google.registry.model.domain.DomainBase;
+import google.registry.model.host.HostResource;
+import google.registry.model.rde.RdeMode;
+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;
+
+/** Loading cache that turns a resource into XML for the various points in time and modes. */
+public class RdeFragmenter {
+  private final Map<WatermarkModePair, Optional<DepositFragment>> cache = new HashMap<>();
+  private final ImmutableMap<DateTime, Supplier<EppResource>> resourceAtTimes;
+  private final RdeMarshaller marshaller;
+
+  long cacheHits = 0;
+  long resourcesNotFound = 0;
+  long resourcesFound = 0;
+
+  public RdeFragmenter(
+      ImmutableMap<DateTime, Supplier<EppResource>> resourceAtTimes, RdeMarshaller marshaller) {
+    this.resourceAtTimes = resourceAtTimes;
+    this.marshaller = marshaller;
+  }
+
+  public Optional<DepositFragment> marshal(DateTime watermark, RdeMode mode) {
+    Optional<DepositFragment> result = cache.get(WatermarkModePair.create(watermark, mode));
+    if (result != null) {
+      cacheHits++;
+      return result;
+    }
+    EppResource resource = resourceAtTimes.get(watermark).get();
+    if (resource == null) {
+      result = Optional.empty();
+      cache.put(WatermarkModePair.create(watermark, RdeMode.FULL), result);
+      cache.put(WatermarkModePair.create(watermark, RdeMode.THIN), result);
+      resourcesNotFound++;
+      return result;
+    }
+    resourcesFound++;
+    if (resource instanceof DomainBase) {
+      result = Optional.of(marshaller.marshalDomain((DomainBase) resource, mode));
+      cache.put(WatermarkModePair.create(watermark, mode), result);
+      return result;
+    } else if (resource instanceof ContactResource) {
+      result = Optional.of(marshaller.marshalContact((ContactResource) resource));
+      cache.put(WatermarkModePair.create(watermark, RdeMode.FULL), result);
+      cache.put(WatermarkModePair.create(watermark, RdeMode.THIN), result);
+      return result;
+    } else if (resource instanceof HostResource) {
+      HostResource host = (HostResource) resource;
+      result =
+          Optional.of(
+              host.isSubordinate()
+                  ? marshaller.marshalSubordinateHost(
+                      host,
+                      // Note that loadAtPointInTime() does cloneProjectedAtTime(watermark) for
+                      // us.
+                      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);
+      return result;
+    } else {
+      throw new IllegalStateException(
+          String.format(
+              "Resource %s of type %s cannot be converted to XML.",
+              resource, resource.getClass().getSimpleName()));
+    }
+  }
+
+  /** Map key for {@link RdeFragmenter} cache. */
+  @AutoValue
+  abstract static class WatermarkModePair {
+    abstract DateTime watermark();
+
+    abstract RdeMode mode();
+
+    static WatermarkModePair create(DateTime watermark, RdeMode mode) {
+      return new AutoValue_RdeFragmenter_WatermarkModePair(watermark, mode);
+    }
+  }
+}
@@ -16,12 +16,10 @@ 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;
-import com.google.auto.value.AutoValue;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.ImmutableSetMultimap;
@@ -34,9 +32,6 @@ import google.registry.model.host.HostResource;
 import google.registry.model.rde.RdeMode;
 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;
@@ -129,7 +124,7 @@ public final class RdeStagingMapper extends Mapper<EppResource, PendingDeposit,
        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);
+    RdeFragmenter fragmenter = new RdeFragmenter(resourceAtTimes, marshaller);

    // Emit resource as an XML fragment for all TLDs and modes pending deposit.
    long resourcesEmitted = 0;
@@ -157,74 +152,4 @@ public final class RdeStagingMapper extends Mapper<EppResource, PendingDeposit,
    // Avoid running out of memory.
    tm().clearSessionCache();
  }
-
-  /** 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, Supplier<EppResource>> resourceAtTimes;
-
-    long cacheHits = 0;
-    long resourcesNotFound = 0;
-    long resourcesFound = 0;
-
-    Fragmenter(ImmutableMap<DateTime, Supplier<EppResource>> resourceAtTimes) {
-      this.resourceAtTimes = resourceAtTimes;
-    }
-
-    Optional<DepositFragment> marshal(DateTime watermark, RdeMode mode) {
-      Optional<DepositFragment> result = cache.get(WatermarkModePair.create(watermark, mode));
-      if (result != null) {
-        cacheHits++;
-        return result;
-      }
-      EppResource resource = resourceAtTimes.get(watermark).get();
-      if (resource == null) {
-        result = Optional.empty();
-        cache.put(WatermarkModePair.create(watermark, RdeMode.FULL), result);
-        cache.put(WatermarkModePair.create(watermark, RdeMode.THIN), result);
-        resourcesNotFound++;
-        return result;
-      }
-      resourcesFound++;
-      if (resource instanceof DomainBase) {
-        result = Optional.of(marshaller.marshalDomain((DomainBase) resource, mode));
-        cache.put(WatermarkModePair.create(watermark, mode), result);
-        return result;
-      } else if (resource instanceof ContactResource) {
-        result = Optional.of(marshaller.marshalContact((ContactResource) resource));
-        cache.put(WatermarkModePair.create(watermark, RdeMode.FULL), result);
-        cache.put(WatermarkModePair.create(watermark, RdeMode.THIN), result);
-        return result;
-      } else if (resource instanceof HostResource) {
-        HostResource host = (HostResource) resource;
-        result =
-            Optional.of(
-                host.isSubordinate()
-                    ? marshaller.marshalSubordinateHost(
-                        host,
-                        // Note that loadAtPointInTime() does cloneProjectedAtTime(watermark) for
-                        // us.
-                        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);
-        return result;
-      } else {
-        throw new AssertionError(resource.toString());
-      }
-    }
-  }
-
-  /** Map key for {@link Fragmenter} cache. */
-  @AutoValue
-  abstract static class WatermarkModePair {
-    abstract DateTime watermark();
-    abstract RdeMode mode();
-
-    static WatermarkModePair create(DateTime watermark, RdeMode mode) {
-      return new AutoValue_RdeStagingMapper_WatermarkModePair(watermark, mode);
-    }
-  }
 }
@@ -42,7 +42,8 @@ import javax.persistence.NoResultException;
    method = GET,
    automaticallyPrintOk = true,
    auth = Auth.AUTH_INTERNAL_OR_ADMIN)
-class ReplicateToDatastoreAction implements Runnable {
+@VisibleForTesting
+public class ReplicateToDatastoreAction implements Runnable {
  public static final String PATH = "/_dr/cron/replicateToDatastore";
  private static final FluentLogger logger = FluentLogger.forEnclosingClass();

@@ -61,12 +62,12 @@ class ReplicateToDatastoreAction implements Runnable {
  }

  @VisibleForTesting
-  List<TransactionEntity> getTransactionBatch() {
+  public List<TransactionEntity> getTransactionBatch() {
    // Get the next batch of transactions that we haven't replicated.
    LastSqlTransaction lastSqlTxnBeforeBatch = ofyTm().transact(() -> LastSqlTransaction.load());
    try {
      return jpaTm()
-          .transact(
+          .transactWithoutBackup(
              () ->
                  jpaTm()
                      .query(
@@ -86,7 +87,7 @@ class ReplicateToDatastoreAction implements Runnable {
   * be aborted.
   */
  @VisibleForTesting
-  boolean applyTransaction(TransactionEntity txnEntity) {
+  public boolean applyTransaction(TransactionEntity txnEntity) {
    logger.atInfo().log("Applying a single transaction Cloud SQL -> Cloud Datastore");
    return ofyTm()
        .transact(
@@ -17,7 +17,7 @@ package google.registry.tools;
 import com.beust.jcommander.Parameter;
 import com.beust.jcommander.Parameters;
 import com.google.template.soy.data.SoyMapData;
-import google.registry.tools.soy.DeleteDomainSoyInfo;
+import google.registry.tools.soy.DomainDeleteSoyInfo;

 /** A command to delete a domain via EPP. */
@Parameters(separators = " =", commandDescription = "Delete domain")
@@ -60,7 +60,7 @@ final class DeleteDomainCommand extends MutatingEppToolCommand {
      // Immediate deletion is accomplished using the superuser extension.
      superuser = true;
    }
-    setSoyTemplate(DeleteDomainSoyInfo.getInstance(), DeleteDomainSoyInfo.DELETEDOMAIN);
+    setSoyTemplate(DomainDeleteSoyInfo.getInstance(), DomainDeleteSoyInfo.DELETEDOMAIN);
    addSoyRecord(clientId, new SoyMapData(
        "domainName", domainName,
        "immediately", immediately,
@@ -17,7 +17,7 @@ package google.registry.tools;
 import com.beust.jcommander.Parameter;
 import com.beust.jcommander.Parameters;
 import com.google.template.soy.data.SoyMapData;
-import google.registry.tools.soy.DeleteHostSoyInfo;
+import google.registry.tools.soy.HostDeleteSoyInfo;

 /** A command to delete a host via EPP. */
@Parameters(separators = " =", commandDescription = "Delete host")
@@ -49,7 +49,7 @@ final class DeleteHostCommand extends MutatingEppToolCommand {

  @Override
  protected void initMutatingEppToolCommand() {
-    setSoyTemplate(DeleteHostSoyInfo.getInstance(), DeleteHostSoyInfo.DELETEHOST);
+    setSoyTemplate(HostDeleteSoyInfo.getInstance(), HostDeleteSoyInfo.DELETEHOST);
    addSoyRecord(clientId, new SoyMapData(
        "hostName", hostName,
        "reason", reason,
@@ -25,7 +25,7 @@ import com.beust.jcommander.Parameters;
 import com.google.common.base.Joiner;
 import com.google.template.soy.data.SoyMapData;
 import google.registry.model.domain.DomainBase;
-import google.registry.tools.soy.RenewDomainSoyInfo;
+import google.registry.tools.soy.DomainRenewSoyInfo;
 import google.registry.util.Clock;
 import java.util.List;
 import java.util.Optional;
@@ -68,7 +68,7 @@ final class RenewDomainCommand extends MutatingEppToolCommand {
      Optional<DomainBase> domainOptional =
          loadByForeignKey(DomainBase.class, domainName, now);
      checkArgumentPresent(domainOptional, "Domain '%s' does not exist or is deleted", domainName);
-      setSoyTemplate(RenewDomainSoyInfo.getInstance(), RenewDomainSoyInfo.RENEWDOMAIN);
+      setSoyTemplate(DomainRenewSoyInfo.getInstance(), DomainRenewSoyInfo.RENEWDOMAIN);
      DomainBase domain = domainOptional.get();
      addSoyRecord(
          isNullOrEmpty(clientId) ? domain.getCurrentSponsorClientId() : clientId,
@@ -0,0 +1,23 @@
+{
+  "name": "RDE/BRDA Deposit Generation",
+  "description": "An Apache Beam pipeline generates RDE or BRDA deposits and deposits them to GCS with GhostRyde encryption.",
+  "parameters": [
+    {
+      "name": "pendings",
+      "label": "The pendings deposits to generate.",
+      "helpText": "A TLD to PendingDeposit map that is serialized and Base64 URL-safe encoded.",
+      "regexes": [
+        "A-Za-z0-9\\-_"
+      ]
+    },
+    {
+      "name": "validationMode",
+      "label": "How strict the marshaller validates the given EPP resources.",
+      "helpText": "If set to LENIENT the marshaller will not warn about missing data on the EPP resources.",
+      "is_optional": true,
+      "regexes": [
+        "^STRICT|LENIENT$"
+      ]
+    }
+  ]
+}
@@ -208,6 +208,30 @@ public class ReplayCommitLogsToSqlActionTest {
    assertExpectedIds("previous to keep");
  }

+  @Test
+  void testReplay_dryRun() throws Exception {
+    action.dryRun = true;
+    DateTime now = fakeClock.nowUtc();
+    jpaTm().transact(() -> jpaTm().insertWithoutBackup(TestObject.create("previous to keep")));
+    Key<CommitLogBucket> bucketKey = getBucketKey(1);
+    Key<CommitLogManifest> manifestKey = CommitLogManifest.createKey(bucketKey, now);
+    saveDiffFileNotToRestore(gcsService, now.minusMinutes(2));
+    jpaTm().transact(() -> SqlReplayCheckpoint.set(now.minusMinutes(1).minusMillis(1)));
+    saveDiffFile(
+        gcsService,
+        createCheckpoint(now.minusMinutes(1)),
+        CommitLogManifest.create(bucketKey, now, null),
+        CommitLogMutation.create(manifestKey, TestObject.create("a")),
+        CommitLogMutation.create(manifestKey, TestObject.create("b")));
+
+    action.run();
+    assertThat(response.getStatus()).isEqualTo(SC_OK);
+    assertThat(response.getPayload())
+        .isEqualTo(
+            "Running in dry-run mode; would have processed %d files. They are (limit 10):\n"
+                + "commit_diff_until_1999-12-31T23:59:00.000Z");
+  }
+
  @Test
  void testReplay_manifestWithNoDeletions() throws Exception {
    DateTime now = fakeClock.nowUtc();
@@ -494,6 +518,8 @@ public class ReplayCommitLogsToSqlActionTest {
  private void runAndAssertSuccess(DateTime expectedCheckpointTime) {
    action.run();
    assertThat(response.getStatus()).isEqualTo(SC_OK);
+    assertThat(response.getPayload())
+        .isEqualTo("ReplayCommitLogsToSqlAction completed successfully.");
    assertThat(jpaTm().transact(SqlReplayCheckpoint::get)).isEqualTo(expectedCheckpointTime);
  }

@@ -0,0 +1,257 @@
+// 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.beam.rde;
+
+import static com.google.common.base.Preconditions.checkState;
+import static com.google.common.collect.ImmutableSet.toImmutableSet;
+import static com.google.common.truth.Truth.assertThat;
+import static google.registry.beam.rde.RdePipeline.decodePendings;
+import static google.registry.beam.rde.RdePipeline.encodePendings;
+import static google.registry.model.common.Cursor.CursorType.RDE_STAGING;
+import static google.registry.model.rde.RdeMode.FULL;
+import static google.registry.model.rde.RdeMode.THIN;
+import static google.registry.persistence.transaction.TransactionManagerFactory.jpaTm;
+import static google.registry.persistence.transaction.TransactionManagerFactory.setTm;
+import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
+import static google.registry.rde.RdeResourceType.CONTACT;
+import static google.registry.rde.RdeResourceType.DOMAIN;
+import static google.registry.rde.RdeResourceType.HOST;
+import static google.registry.rde.RdeResourceType.REGISTRAR;
+import static google.registry.testing.AppEngineExtension.loadInitialData;
+import static google.registry.testing.DatabaseHelper.createTld;
+import static google.registry.testing.DatabaseHelper.newHostResource;
+import static google.registry.testing.DatabaseHelper.persistActiveContact;
+import static google.registry.testing.DatabaseHelper.persistActiveDomain;
+import static google.registry.testing.DatabaseHelper.persistDeletedDomain;
+import static google.registry.testing.DatabaseHelper.persistEppResource;
+import static google.registry.testing.DatabaseHelper.persistNewRegistrar;
+import static org.joda.time.Duration.standardDays;
+
+import com.google.common.base.Splitter;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSetMultimap;
+import com.google.common.collect.Iterables;
+import com.google.common.collect.Streams;
+import google.registry.beam.TestPipelineExtension;
+import google.registry.model.host.HostResource;
+import google.registry.model.registrar.Registrar;
+import google.registry.model.registrar.Registrar.State;
+import google.registry.persistence.transaction.JpaTestRules;
+import google.registry.persistence.transaction.JpaTestRules.JpaIntegrationTestExtension;
+import google.registry.persistence.transaction.TransactionManager;
+import google.registry.rde.DepositFragment;
+import google.registry.rde.PendingDeposit;
+import google.registry.rde.RdeResourceType;
+import google.registry.testing.DatastoreEntityExtension;
+import google.registry.testing.FakeClock;
+import java.util.function.Function;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import java.util.stream.Stream;
+import org.apache.beam.sdk.options.PipelineOptionsFactory;
+import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.transforms.GroupByKey;
+import org.apache.beam.sdk.values.KV;
+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.Order;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+/** Unit tests for {@link RdePipeline}. */
+public class RdePipelineTest {
+
+  private static final String REGISTRAR_NAME_PATTERN =
+      "<rdeRegistrar:name>(.*)</rdeRegistrar:name>";
+
+  private static final String DOMAIN_NAME_PATTERN = "<rdeDomain:name>(.*)</rdeDomain:name>";
+
+  private static final String CONTACT_ID_PATTERN = "<rdeContact:id>(.*)</rdeContact:id>";
+
+  private static final String HOST_NAME_PATTERN = "<rdeHost:name>(.*)</rdeHost:name>";
+
+  private static final ImmutableSetMultimap<String, PendingDeposit> PENDINGS =
+      ImmutableSetMultimap.of(
+          "pal",
+          PendingDeposit.create(
+              "pal", DateTime.parse("2000-01-01TZ"), FULL, RDE_STAGING, standardDays(1)),
+          "pal",
+          PendingDeposit.create(
+              "pal", DateTime.parse("2000-01-01TZ"), THIN, RDE_STAGING, standardDays(1)),
+          "fun",
+          PendingDeposit.create(
+              "fun", DateTime.parse("2000-01-01TZ"), FULL, RDE_STAGING, standardDays(1)));
+
+  // This is the default creation time for test data.
+  private final FakeClock clock = new FakeClock(DateTime.parse("1999-12-31TZ"));
+
+  // The pipeline runs in a different thread, which needs to be masqueraded as a GAE thread as well.
+  @RegisterExtension
+  @Order(Order.DEFAULT - 1)
+  final DatastoreEntityExtension datastore = new DatastoreEntityExtension().allThreads(true);
+
+  @RegisterExtension
+  final JpaIntegrationTestExtension database =
+      new JpaTestRules.Builder().withClock(clock).buildIntegrationTestRule();
+
+  @RegisterExtension
+  final TestPipelineExtension pipeline =
+      TestPipelineExtension.create().enableAbandonedNodeEnforcement(true);
+
+  private final RdePipelineOptions options =
+      PipelineOptionsFactory.create().as(RdePipelineOptions.class);
+
+  private RdePipeline rdePipeline;
+
+  private TransactionManager originalTm;
+
+  @BeforeEach
+  void beforeEach() throws Exception {
+    originalTm = tm();
+    setTm(jpaTm());
+    loadInitialData();
+
+    // Two real registrars have been created by loadInitialData(), named "New Registrar" and "The
+    // Registrar". Create one included registrar (external_monitoring) and two excluded ones.
+    Registrar monitoringRegistrar =
+        persistNewRegistrar("monitoring", "monitoring", Registrar.Type.MONITORING, null);
+    Registrar testRegistrar = persistNewRegistrar("test", "test", Registrar.Type.TEST, null);
+    Registrar externalMonitoringRegistrar =
+        persistNewRegistrar(
+            "externalmonitor", "external_monitoring", Registrar.Type.EXTERNAL_MONITORING, 9997L);
+    // Set Registrar states which are required for reporting.
+    tm().transact(
+            () ->
+                tm().putAll(
+                        ImmutableList.of(
+                            externalMonitoringRegistrar.asBuilder().setState(State.ACTIVE).build(),
+                            testRegistrar.asBuilder().setState(State.ACTIVE).build(),
+                            monitoringRegistrar.asBuilder().setState(State.ACTIVE).build())));
+
+    createTld("pal");
+    createTld("fun");
+    createTld("cat");
+
+    // Also persists a "contact1234" contact in the process.
+    persistActiveDomain("hello.pal");
+    persistActiveDomain("kitty.fun");
+    // Should not appear anywhere.
+    persistActiveDomain("lol.cat");
+    persistDeletedDomain("deleted.pal", DateTime.parse("1999-12-30TZ"));
+
+    persistActiveContact("contact456");
+
+    HostResource host = persistEppResource(newHostResource("old.host.test"));
+    // Set the clock to 2000-01-02, the updated host should NOT show up in RDE.
+    clock.advanceBy(Duration.standardDays(2));
+    persistEppResource(host.asBuilder().setHostName("new.host.test").build());
+
+    options.setPendings(encodePendings(PENDINGS));
+    // The EPP resources created in tests do not have all the fields populated, using a STRICT
+    // validation mode will result in a lot of warnings during marshalling.
+    options.setValidationMode("LENIENT");
+    rdePipeline = new RdePipeline(options);
+  }
+
+  @AfterEach
+  void afterEach() {
+    setTm(originalTm);
+  }
+
+  @Test
+  void testSuccess_encodeAndDecodePendingsMap() throws Exception {
+    String encodedString = encodePendings(PENDINGS);
+    assertThat(decodePendings(encodedString)).isEqualTo(PENDINGS);
+  }
+
+  @Test
+  void testSuccess_createFragments() {
+    PAssert.that(
+            rdePipeline
+                .createFragments(pipeline)
+                .apply("Group by PendingDeposit", GroupByKey.create()))
+        .satisfies(
+            kvs -> {
+              kvs.forEach(
+                  kv -> {
+                    // Registrar fragments.
+                    assertThat(
+                            getFragmentForType(kv, REGISTRAR)
+                                .map(getXmlElement(REGISTRAR_NAME_PATTERN))
+                                .collect(toImmutableSet()))
+                        // The same registrars are attached to all the pending deposits.
+                        .containsExactly("New Registrar", "The Registrar", "external_monitoring");
+                    // Domain fragments.
+                    assertThat(
+                            getFragmentForType(kv, DOMAIN)
+                                .map(getXmlElement(DOMAIN_NAME_PATTERN))
+                                .anyMatch(
+                                    domain ->
+                                        // Deleted domain should not be included
+                                        domain.equals("deleted.pal")
+                                            // Only domains on the pending deposit's tld should
+                                            // appear.
+                                            || !kv.getKey()
+                                                .tld()
+                                                .equals(
+                                                    Iterables.get(
+                                                        Splitter.on('.').split(domain), 1))))
+                        .isFalse();
+                    if (kv.getKey().mode().equals(FULL)) {
+                      // Contact fragments.
+                      assertThat(
+                              getFragmentForType(kv, CONTACT)
+                                  .map(getXmlElement(CONTACT_ID_PATTERN))
+                                  .collect(toImmutableSet()))
+                          // The same contacts are attached too all pending deposits.
+                          .containsExactly("contact1234", "contact456");
+                      // Host fragments.
+                      assertThat(
+                              getFragmentForType(kv, HOST)
+                                  .map(getXmlElement(HOST_NAME_PATTERN))
+                                  .collect(toImmutableSet()))
+                          // Should load the resource before update.
+                          .containsExactly("old.host.test");
+                    } else {
+                      // BRDA does not contain contact or hosts.
+                      assertThat(
+                              Streams.stream(kv.getValue())
+                                  .anyMatch(
+                                      fragment ->
+                                          fragment.type().equals(CONTACT)
+                                              || fragment.type().equals(HOST)))
+                          .isFalse();
+                    }
+                  });
+              return null;
+            });
+    pipeline.run().waitUntilFinish();
+  }
+
+  private static Function<DepositFragment, String> getXmlElement(String pattern) {
+    return (fragment) -> {
+      Matcher matcher = Pattern.compile(pattern).matcher(fragment.xml());
+      checkState(matcher.find(), "Missing %s in xml.", pattern);
+      return matcher.group(1);
+    };
+  }
+
+  private static Stream<DepositFragment> getFragmentForType(
+      KV<PendingDeposit, Iterable<DepositFragment>> kv, RdeResourceType type) {
+    return Streams.stream(kv.getValue()).filter(fragment -> fragment.type().equals(type));
+  }
+}
@@ -145,7 +145,7 @@ class EppPointInTimeTest {
    tm().clearSessionCache();
    assertAboutImmutableObjects()
        .that(loadAtPointInTime(latest, timeAtCreate.plusDays(1)))
-        .hasFieldsEqualTo(domainAfterCreate);
+        .isEqualExceptFields(domainAfterCreate, "updateTimestamp");

    tm().clearSessionCache();
    if (tm().isOfy()) {
@@ -159,18 +159,18 @@ class EppPointInTimeTest {
      // second update occurred one millisecond later.
      assertAboutImmutableObjects()
          .that(loadAtPointInTime(latest, timeAtFirstUpdate))
-          .hasFieldsEqualTo(domainAfterFirstUpdate);
+          .isEqualExceptFields(domainAfterFirstUpdate, "updateTimestamp");
    }

    tm().clearSessionCache();
    assertAboutImmutableObjects()
        .that(loadAtPointInTime(latest, timeAtSecondUpdate))
-        .hasFieldsEqualTo(domainAfterSecondUpdate);
+        .isEqualExceptFields(domainAfterSecondUpdate, "updateTimestamp");

    tm().clearSessionCache();
    assertAboutImmutableObjects()
        .that(loadAtPointInTime(latest, timeAtSecondUpdate.plusDays(1)))
-        .hasFieldsEqualTo(domainAfterSecondUpdate);
+        .isEqualExceptFields(domainAfterSecondUpdate, "updateTimestamp");

    // Deletion time has millisecond granularity due to isActive() check.
    tm().clearSessionCache();
@@ -51,7 +51,7 @@ class HostInfoFlowTest extends ResourceFlowTestCase<HostInfoFlow, HostResource>

  @Order(value = Order.DEFAULT - 2)
  @RegisterExtension
-  final ReplayExtension replayExtension = ReplayExtension.createWithCompare(clock);
+  final ReplayExtension replayExtension = ReplayExtension.createWithDoubleReplay(clock);

  HostInfoFlowTest() {
    setEppInput("host_info.xml", ImmutableMap.of("HOSTNAME", "ns1.example.tld"));
@@ -122,6 +122,13 @@ public class ContactResourceTest extends EntityTestCase {
    contactResource = persistResource(cloneAndSetAutoTimestamps(originalContact));
  }

+  @Test
+  void testContactBaseToContactResource() {
+    ImmutableObjectSubject.assertAboutImmutableObjects()
+        .that(new ContactResource.Builder().copyFrom(contactResource).build())
+        .isEqualExceptFields(contactResource, "updateTimestamp", "revisions");
+  }
+
  @Test
  void testCloudSqlPersistence_failWhenViolateForeignKeyConstraint() {
    assertThrowForeignKeyViolation(() -> jpaTm().transact(() -> jpaTm().insert(originalContact)));
@@ -39,6 +39,7 @@ import com.google.common.collect.Streams;
 import com.googlecode.objectify.Key;
 import google.registry.model.EntityTestCase;
 import google.registry.model.ImmutableObject;
+import google.registry.model.ImmutableObjectSubject;
 import google.registry.model.billing.BillingEvent;
 import google.registry.model.billing.BillingEvent.Reason;
 import google.registry.model.contact.ContactResource;
@@ -171,6 +172,13 @@ public class DomainBaseTest extends EntityTestCase {
                    .build()));
  }

+  @Test
+  void testDomainContentToDomainBase() {
+    ImmutableObjectSubject.assertAboutImmutableObjects()
+        .that(new DomainBase.Builder().copyFrom(domain).build())
+        .isEqualExceptFields(domain, "updateTimestamp", "revisions");
+  }
+
  @Test
  void testPersistence() {
    // Note that this only verifies that the value stored under the foreign key is the same as that
@@ -31,6 +31,7 @@ import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableSet;
 import com.google.common.net.InetAddresses;
 import google.registry.model.EntityTestCase;
+import google.registry.model.ImmutableObjectSubject;
 import google.registry.model.domain.DomainBase;
 import google.registry.model.eppcommon.StatusValue;
 import google.registry.model.eppcommon.Trid;
@@ -89,6 +90,13 @@ class HostResourceTest extends EntityTestCase {
                    .build()));
  }

+  @TestOfyAndSql
+  void testHostBaseToHostResource() {
+    ImmutableObjectSubject.assertAboutImmutableObjects()
+        .that(new HostResource.Builder().copyFrom(host).build())
+        .isEqualExceptFields(host, "updateTimestamp", "revisions");
+  }
+
  @TestOfyAndSql
  void testPersistence() {
    HostResource newHost = host.asBuilder().setRepoId("NEWHOST").build();
@@ -46,6 +46,7 @@ import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junitpioneer.jupiter.RetryingTest;

 public class ReplicateToDatastoreActionTest {

@@ -97,7 +98,7 @@ public class ReplicateToDatastoreActionTest {
    RegistryConfig.overrideCloudSqlReplicateTransactions(false);
  }

-  @Test
+  @RetryingTest(4)
  public void testReplication() {
    TestEntity foo = new TestEntity("foo");
    TestEntity bar = new TestEntity("bar");
@@ -127,7 +128,7 @@ public class ReplicateToDatastoreActionTest {
    assertThat(ofyTm().transact(() -> ofyTm().loadByKey(baz.key()))).isEqualTo(baz);
  }

-  @Test
+  @RetryingTest(4)
  public void testReplayFromLastTxn() {
    TestEntity foo = new TestEntity("foo");
    TestEntity bar = new TestEntity("bar");
@@ -149,7 +150,7 @@ public class ReplicateToDatastoreActionTest {
    assertThat(ofyTm().transact(() -> ofyTm().loadByKeyIfPresent(foo.key()).isPresent())).isFalse();
  }

-  @Test
+  @RetryingTest(4)
  public void testUnintentionalConcurrency() {
    TestEntity foo = new TestEntity("foo");
    TestEntity bar = new TestEntity("bar");
@@ -184,7 +185,7 @@ public class ReplicateToDatastoreActionTest {
            Level.WARNING, "Ignoring transaction 1, which appears to have already been applied.");
  }

-  @Test
+  @RetryingTest(4)
  public void testMissingTransactions() {
    // Write a transaction (should have a transaction id of 1).
    TestEntity foo = new TestEntity("foo");
@@ -477,7 +477,7 @@ public final class AppEngineExtension implements BeforeEachCallback, AfterEachCa
  public void afterEach(ExtensionContext context) throws Exception {
    checkArgumentNotNull(context, "The ExtensionContext must not be null");
    try {
-      // If there is a replay extension, we'll want to call its replayToSql() method.
+      // If there is a replay extension, we'll want to call its replay() method.
      //
      // We have to provide this hook here for ReplayExtension instead of relying on
      // ReplayExtension's afterEach() method because of ordering and the conflation of environment
@@ -492,7 +492,7 @@ public final class AppEngineExtension implements BeforeEachCallback, AfterEachCa
          (ReplayExtension)
              context.getStore(ExtensionContext.Namespace.GLOBAL).get(ReplayExtension.class);
      if (replayer != null) {
-        replayer.replayToSql();
+        replayer.replay();
      }

      if (withCloudSql) {
@@ -1085,8 +1085,7 @@ public class DatabaseHelper {
                          .setClientId(resource.getCreationClientId())
                          .setType(getHistoryEntryType(resource))
                          .setModificationTime(tm().getTransactionTime())
-                          .build()
-                          .toChildHistoryEntity());
+                          .build());
              ofyTmOrDoNothing(
                  () -> tm().put(ForeignKeyIndex.create(resource, resource.getDeletionTime())));
            });
@@ -18,6 +18,8 @@ import static google.registry.model.ofy.ObjectifyService.auditedOfy;

 import com.google.apphosting.api.ApiProxy;
 import com.google.apphosting.api.ApiProxy.Environment;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
 import java.util.Map;
 import org.junit.jupiter.api.extension.AfterEachCallback;
 import org.junit.jupiter.api.extension.BeforeEachCallback;
@@ -25,7 +27,7 @@ import org.junit.jupiter.api.extension.ExtensionContext;
 import org.testcontainers.shaded.com.google.common.collect.ImmutableMap;

 /**
- * Allows instantiation of Datastore {@code Entity entities} without the heavyweight {@code
+ * Allows instantiation of Datastore {@code Entity}s without the heavyweight {@link
 * AppEngineExtension}.
 *
 * <p>When an Ofy key is created, by calling the various Key.create() methods, whether the current
@@ -44,6 +46,27 @@ public class DatastoreEntityExtension implements BeforeEachCallback, AfterEachCa

  private static final Environment PLACEHOLDER_ENV = new PlaceholderEnvironment();

+  private boolean allThreads = false;
+
+  /**
+   * Whether all threads should be masqueraded as GAE threads.
+   *
+   * <p>This is particularly useful when new threads are spawned during a test. For example when
+   * testing Beam pipelines, the test pipeline runs the transforms in separate threads than the test
+   * thread. If Ofy keys are created in transforms, this value needs to be set to true.
+   *
+   * <p>Warning: by setting this value to true, any thread spawned by the current JVM will be have
+   * the thread local property set to the placeholder value during the execution of the current
+   * test, including those running other tests in parallel. This may or may not cause an issue when
+   * other tests have {@link AppEngineExtension} registered, that creates a much more fully
+   * functional GAE test environment. Consider moving tests using this extension to {@code
+   * outcastTest} if necessary.
+   */
+  public DatastoreEntityExtension allThreads(boolean value) {
+    allThreads = value;
+    return this;
+  }
+
  @Override
  public void beforeEach(ExtensionContext context) {
    ApiProxy.setEnvironmentForCurrentThread(PLACEHOLDER_ENV);
@@ -51,12 +74,21 @@ public class DatastoreEntityExtension implements BeforeEachCallback, AfterEachCa
    // will load the ObjectifyService class, whose static initialization block registers all Ofy
    // entities.
    auditedOfy();
+    if (allThreads) {
+      ApiProxy.setEnvironmentFactory(() -> PLACEHOLDER_ENV);
+    }
  }

  @Override
-  public void afterEach(ExtensionContext context) {
+  public void afterEach(ExtensionContext context)
+      throws NoSuchMethodException, InvocationTargetException, IllegalAccessException {
    // Clear the cached instance.
-    ApiProxy.setEnvironmentForCurrentThread(null);
+    ApiProxy.clearEnvironmentForCurrentThread();
+    if (allThreads) {
+      Method method = ApiProxy.class.getDeclaredMethod("clearEnvironmentFactory");
+      method.setAccessible(true);
+      method.invoke(null);
+    }
  }

  private static final class PlaceholderEnvironment implements Environment {
@@ -51,15 +51,23 @@ class DualDatabaseTestInvocationContextProvider implements TestTemplateInvocatio
    return true;
  }

+  /**
+   * Returns true if "context" is an objectify unit test.
+   *
+   * <p>Provided to allow ReplayExtension to make this determination.
+   */
+  static boolean inOfyContext(ExtensionContext context) {
+    return (DatabaseType) context.getStore(NAMESPACE).get(INJECTED_TM_SUPPLIER_KEY)
+        == DatabaseType.OFY;
+  }
+
  @Override
  public Stream<TestTemplateInvocationContext> provideTestTemplateInvocationContexts(
      ExtensionContext context) {
    TestTemplateInvocationContext ofyContext =
-        createInvocationContext(
-            context.getDisplayName() + " with Datastore", TransactionManagerFactory::ofyTm);
+        createInvocationContext(context.getDisplayName() + " with Datastore", DatabaseType.OFY);
    TestTemplateInvocationContext sqlContext =
-        createInvocationContext(
-            context.getDisplayName() + " with PostgreSQL", TransactionManagerFactory::jpaTm);
+        createInvocationContext(context.getDisplayName() + " with PostgreSQL", DatabaseType.JPA);
    Method testMethod = context.getTestMethod().orElseThrow(IllegalStateException::new);
    if (testMethod.isAnnotationPresent(TestOfyAndSql.class)) {
      return Stream.of(ofyContext, sqlContext);
@@ -74,7 +82,7 @@ class DualDatabaseTestInvocationContextProvider implements TestTemplateInvocatio
  }

  private TestTemplateInvocationContext createInvocationContext(
-      String name, Supplier<? extends TransactionManager> tmSupplier) {
+      String name, DatabaseType databaseType) {
    return new TestTemplateInvocationContext() {
      @Override
      public String getDisplayName(int invocationIndex) {
@@ -83,17 +91,17 @@ class DualDatabaseTestInvocationContextProvider implements TestTemplateInvocatio

      @Override
      public List<Extension> getAdditionalExtensions() {
-        return ImmutableList.of(new DatabaseSwitchInvocationContext(tmSupplier));
+        return ImmutableList.of(new DatabaseSwitchInvocationContext(databaseType));
      }
    };
  }

  private static class DatabaseSwitchInvocationContext implements TestInstancePostProcessor {

-    private Supplier<? extends TransactionManager> tmSupplier;
+    private DatabaseType databaseType;

-    private DatabaseSwitchInvocationContext(Supplier<? extends TransactionManager> tmSupplier) {
-      this.tmSupplier = tmSupplier;
+    private DatabaseSwitchInvocationContext(DatabaseType databaseType) {
+      this.databaseType = databaseType;
    }

    @Override
@@ -113,7 +121,7 @@ class DualDatabaseTestInvocationContextProvider implements TestTemplateInvocatio
        throw new IllegalStateException(
            "AppEngineExtension in @DualDatabaseTest test must set withDatastoreAndCloudSql()");
      }
-      context.getStore(NAMESPACE).put(INJECTED_TM_SUPPLIER_KEY, tmSupplier);
+      context.getStore(NAMESPACE).put(INJECTED_TM_SUPPLIER_KEY, databaseType);
    }

    private static ImmutableList<Field> getAppEngineExtensionFields(Class<?> clazz) {
@@ -144,10 +152,9 @@ class DualDatabaseTestInvocationContextProvider implements TestTemplateInvocatio
                }
              });
      context.getStore(NAMESPACE).put(ORIGINAL_TM_KEY, tm());
-      Supplier<? extends TransactionManager> tmSupplier =
-          (Supplier<? extends TransactionManager>)
-              context.getStore(NAMESPACE).get(INJECTED_TM_SUPPLIER_KEY);
-      TransactionManagerFactory.setTm(tmSupplier.get());
+      DatabaseType databaseType =
+          (DatabaseType) context.getStore(NAMESPACE).get(INJECTED_TM_SUPPLIER_KEY);
+      TransactionManagerFactory.setTm(databaseType.getTm());
    }
  }

@@ -171,4 +178,20 @@ class DualDatabaseTestInvocationContextProvider implements TestTemplateInvocatio
    return testInstance.getClass().isAnnotationPresent(DualDatabaseTest.class)
        && isDeclaredTestMethod;
  }
+
+  private enum DatabaseType {
+    JPA(TransactionManagerFactory::jpaTm),
+    OFY(TransactionManagerFactory::ofyTm);
+
+    @SuppressWarnings("Immutable") // Supplier is immutable, but not annotated as such.
+    private final Supplier<? extends TransactionManager> supplier;
+
+    DatabaseType(Supplier<? extends TransactionManager> supplier) {
+      this.supplier = supplier;
+    }
+
+    TransactionManager getTm() {
+      return supplier.get();
+    }
+  }
 }
@@ -22,13 +22,17 @@ import com.google.common.base.Suppliers;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
 import com.googlecode.objectify.Key;
+import google.registry.config.RegistryConfig;
 import google.registry.model.ImmutableObject;
 import google.registry.model.ofy.CommitLogBucket;
 import google.registry.model.ofy.ReplayQueue;
 import google.registry.model.ofy.TransactionInfo;
 import google.registry.persistence.VKey;
+import google.registry.persistence.transaction.TransactionEntity;
 import google.registry.schema.replay.DatastoreEntity;
+import google.registry.schema.replay.ReplicateToDatastoreAction;
 import java.util.Optional;
+import javax.annotation.Nullable;
 import org.junit.jupiter.api.extension.AfterEachCallback;
 import org.junit.jupiter.api.extension.BeforeEachCallback;
 import org.junit.jupiter.api.extension.ExtensionContext;
@@ -48,19 +52,27 @@ public class ReplayExtension implements BeforeEachCallback, AfterEachCallback {

  FakeClock clock;
  boolean compare;
+  boolean replayed = false;
+  boolean inOfyContext;
  InjectExtension injectExtension = new InjectExtension();
+  @Nullable ReplicateToDatastoreAction sqlToDsReplicator;

-  private ReplayExtension(FakeClock clock, boolean compare) {
+  private ReplayExtension(
+      FakeClock clock, boolean compare, @Nullable ReplicateToDatastoreAction sqlToDsReplicator) {
    this.clock = clock;
    this.compare = compare;
+    this.sqlToDsReplicator = sqlToDsReplicator;
  }

  public static ReplayExtension createWithCompare(FakeClock clock) {
-    return new ReplayExtension(clock, true);
+    return new ReplayExtension(clock, true, null);
  }

-  public static ReplayExtension createWithoutCompare(FakeClock clock) {
-    return new ReplayExtension(clock, false);
+  /**
+   * Create a replay extension that replays from SQL to cloud datastore when running in SQL mode.
+   */
+  public static ReplayExtension createWithDoubleReplay(FakeClock clock) {
+    return new ReplayExtension(clock, true, new ReplicateToDatastoreAction(clock));
  }

  @Override
@@ -74,16 +86,27 @@ public class ReplayExtension implements BeforeEachCallback, AfterEachCallback {
    DatabaseHelper.setClock(clock);
    DatabaseHelper.setAlwaysSaveWithBackup(true);
    ReplayQueue.clear();
+
+    // When running in JPA mode with double replay enabled, enable JPA transaction replication.
+    // Note that we can't just use isOfy() here because this extension gets run before the dual-test
+    // transaction manager gets injected.
+    inOfyContext = DualDatabaseTestInvocationContextProvider.inOfyContext(context);
+    if (sqlToDsReplicator != null && !inOfyContext) {
+      RegistryConfig.overrideCloudSqlReplicateTransactions(true);
+    }
+
    context.getStore(ExtensionContext.Namespace.GLOBAL).put(ReplayExtension.class, this);
  }

  @Override
  public void afterEach(ExtensionContext context) {
    // This ensures that we do the replay even if we're not called from AppEngineExtension.  It
-    // should be safe to call replayToSql() twice, as the replay queue should be empty the second
-    // time.
-    replayToSql();
+    // is safe to call replay() twice, as the method ensures idempotence.
+    replay();
    injectExtension.afterEach(context);
+    if (sqlToDsReplicator != null) {
+      RegistryConfig.overrideCloudSqlReplicateTransactions(false);
+    }
  }

  private static ImmutableSet<String> NON_REPLICATED_TYPES =
@@ -104,7 +127,26 @@ public class ReplayExtension implements BeforeEachCallback, AfterEachCallback {
          "ForeignKeyContactIndex",
          "ForeignKeyDomainIndex");

-  public void replayToSql() {
+  public void replay() {
+    if (!replayed) {
+      if (inOfyContext) {
+        replayToSql();
+      } else {
+        // Disable database backups.  For unknown reason, if we don't do this we get residual commit
+        // log entries that cause timestamp inversions in other tests.
+        DatabaseHelper.setAlwaysSaveWithBackup(false);
+
+        // Do the ofy replay.
+        replayToOfy();
+
+        // Clean out anything that ends up in the replay queue.
+        ReplayQueue.clear();
+      }
+      replayed = true;
+    }
+  }
+
+  private void replayToSql() {
    DatabaseHelper.setAlwaysSaveWithBackup(false);
    ImmutableMap<Key<?>, Object> changes = ReplayQueue.replay();

@@ -139,4 +181,18 @@ public class ReplayExtension implements BeforeEachCallback, AfterEachCallback {
      }
    }
  }
+
+  private void replayToOfy() {
+    if (sqlToDsReplicator == null) {
+      return;
+    }
+
+    // TODO(mmuller): Verify that all entities are the same across both databases.
+    for (TransactionEntity txn : sqlToDsReplicator.getTransactionBatch()) {
+      if (sqlToDsReplicator.applyTransaction(txn)) {
+        break;
+      }
+      clock.advanceOneMilli();
+    }
+  }
 }
@@ -93,7 +93,9 @@ steps:
      google.registry.beam.spec11.Spec11Pipeline \
      google/registry/beam/spec11_pipeline_metadata.json \
      google.registry.beam.invoicing.InvoicingPipeline \
-      google/registry/beam/invoicing_pipeline_metadata.json
+      google/registry/beam/invoicing_pipeline_metadata.json \
+      google.registry.beam.rde.RdePipeline \
+      google/registry/beam/rde_pipeline_metadata.json
 # Tentatively build and publish Cloud SQL schema jar here, before schema release
 # process is finalized. Also publish nomulus:core jars that are needed for
 # server/schema compatibility tests.
Author	SHA1	Message	Date
gbrodman	e5b9ff1498	Add a dry-run option to commit-log-replay action and use it in Sandbox (#1234 )	2021-07-03 11:18:00 -04:00
Ben McIlwain	3f9fec98d5	Add more logging to replay commit logs action (#1232 )	2021-07-02 18:06:04 -04:00
Ben McIlwain	4e30d020ca	Set payload response in happy path of ReplayCommitLogsToSqlAction (#1229 ) * Set payload response in happy path of ReplayCommitLogsToSqlAction I suspect this may be the reason the logs are missing on the happy path (when it runs successfully), but are visible on the exception paths (which do set the payload response). I don't think App Engine likes it when a Web request terminates without a response. This also adds more logging and error handling.	2021-07-01 18:21:17 -04:00
Lai Jiang	047444831b	Add a Beam pipeline to generate RDE deposit (part 1) (#1219 ) This is the first part of the RdeStagingAction SQL migration where the mapper logic is implemented in Beam. A few helper methods are added to convert the DomainContent, HostBase and ContactBase to their respective terminal child classes. This is necessary and possible because the child classes do not have extra fields and the base classes exist only to be embedded to other entities (such as the various HistoryEntry entities). The conversion is necessary because most of our code expects the terminal classes, such as the RdeMarshaller's various marshallXXX() methods. The alternative would be to change all the call sites, which seems to be much more disruptive. Unfortunately there is is no good way to do this conversion than just creating a builder and setting every fields there is. <!-- Reviewable:start --> --- This change is [<img src="https://reviewable.io/review_button.svg" height="34" align="absmiddle" alt="Reviewable"/>](https://reviewable.io/reviews/google/nomulus/1219) <!-- Reviewable:end -->	2021-06-30 13:54:24 -04:00
Weimin Yu	7adcbee5ad	Retry flaky tests for ReplicateToDatastoreAction (#1226 ) * Retry flaky tests for ReplicateToDatastoreAction The occassional failures seem to be caused by the test Datastore.	2021-06-29 17:12:02 -04:00
Michael Muller	78a750b7e1	Support testing SQL -> DS replication in ReplayExt (#1216 ) * Support testing SQL -> DS replication in ReplayExt Support testing of Postgres -> Datastore replication in the ReplayExtension when running in SQL mode in a DualDatabaseTest. This is currently only enabled for one test (HostInfoFlowTest) since this form of replication is likely to be problematic in many cases. As part of this change: - Add a thread-local flag so that we don't attempt to do certain data transformations when serializing entities for storage in a Transaction record. (These typically need to be called in a datastore transaction). - Replace tm() in datastore translators with ofyTm() (these should only be called from within an ofy transaction) and also in the replay system itself. - Add a transactWithoutBackup() method for use within the replay itself. - Prevent replication of entities that are not intended to be replicated. - Make some of the ReplicateToDatastoreAction methods public so we can invoke them from ReplayExtension. - Change the way that the test type is stored in the extension context in a DualDatabaseTest so that we can check for it from the ReplayExtension. * Limit number of tests and show output Trying to debug why these are failing in kokoro. * Move HostInfoFlowTest to fragile for now The test now manipulates a globel variable that causes problems for other tests. There's likely a better fix for this, but for purposes of this PR we can just move it to "fragile." * Fix a few more problems - "replay" flag should have been initialized to false -- as it stands, replay wasn't happening. - disable "always save with backup" in the datastore helper, we were apparently getting some unwanted commit log entries that were causing timestamp inversions in other tests. Also clear out the replay queue just for good hygiene. - Check for a null replicator in replayToOfy before proceeding. - Use a local inOfyContext flag to track whether we're in ofy context, as the tm() function is less reliable in dual-database tests.	2021-06-29 10:00:39 -04:00
Ben McIlwain	2e8a1c422d	Set HistoryEntry modification time in FlowModule (#1222 ) * Set HistoryEntry modification time in FlowModule Rather than having to set it individually to now (the current transaction time) in every transactional flow, just do it once at the beginning when the HistoryEntry.Builder is first being provided. This is also safer, as just doing it in one place gives us stronger guarantees that it always corresponds to the execution time of the flow, rather than leaving the potential open that in one flow it's unintentionally set to the wrong thing.	2021-06-29 09:05:12 -04:00
gbrodman	0e5605b175	Set a 5min time limit on the SQL replay action (#1224 ) This means we avoid GAE request timeouts and can get progress logs more quickly (logs weren't showing up on GAE in Sandbox).	2021-06-28 17:01:16 -04:00
Ben McIlwain	a10b5d8b30	Rename a few soy files for consistency (#1223 ) * Rename a few soy files for consistency This prefers the ResourceAction.soy naming convention for .soy files that contain EPP XMLs so that they match the name of the corresponding EPP flow. E.g. DomainDelete.soy now matches DomainDeleteFlow.java	2021-06-28 12:00:08 -04:00