Apply registrar code change after registrar schema change is taken effect (#1254)

* Apply code change after registrar schema change is taken effect

SECURITY.md

@@ -0,0 +1,4 @@
+To report a security issue, please use http://g.co/vulnz. We use
+http://g.co/vulnz for our intake, and do coordination and disclosure here on
+GitHub (including using GitHub Security Advisory). The Google Security Team will
+respond within 5 working days of your report on g.co/vulnz.

core/build.gradle

@@ -74,8 +74,6 @@ def fragileTestPatterns = [
     "google/registry/cron/TldFanoutActionTest.*",
     // Test Datastore inexplicably aborts transaction.
     "google/registry/model/tmch/ClaimsListShardTest.*",
-    // Creates large object (64MBytes), occasionally throws OOM error.
-    "google/registry/model/server/KmsSecretRevisionTest.*",
     // Changes cache timeouts and for some reason appears to have contention
     // with other tests.
     "google/registry/whois/WhoisCommandFactoryTest.*",

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

@@ -15,6 +15,7 @@
 package google.registry.backup;
 
 import static com.google.common.base.Preconditions.checkState;
+import static com.google.common.collect.ImmutableList.toImmutableList;
 import static google.registry.backup.BackupUtils.GcsMetadataKeys.LOWER_BOUND_CHECKPOINT;
 import static google.registry.backup.ExportCommitLogDiffAction.DIFF_FILE_PREFIX;
 import static google.registry.util.DateTimeUtils.START_OF_TIME;
@@ -23,13 +24,13 @@ import static google.registry.util.DateTimeUtils.latestOf;
 
 import com.google.cloud.storage.BlobId;
 import com.google.cloud.storage.BlobInfo;
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.ImmutableList;
 import com.google.common.flogger.FluentLogger;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.ListeningExecutorService;
 import com.google.common.util.concurrent.UncheckedExecutionException;
-import dagger.Lazy;
 import google.registry.backup.BackupModule.Backups;
 import google.registry.gcs.GcsUtils;
 import java.io.IOException;
@@ -39,6 +40,7 @@ import java.util.TreeMap;
 import java.util.concurrent.ScheduledExecutorService;
 import javax.annotation.Nullable;
 import javax.inject.Inject;
+import javax.inject.Provider;
 import org.joda.time.DateTime;
 
 /** Utility class to list commit logs diff files stored on GCS. */
@@ -51,7 +53,7 @@ class GcsDiffFileLister {
 
   @Inject GcsUtils gcsUtils;
 
-  @Inject @Backups Lazy<ListeningExecutorService> lazyExecutor;
+  @Inject @Backups Provider<ListeningExecutorService> executorProvider;
   @Inject ScheduledExecutorService scheduledExecutorService;
 
   @Inject
@@ -113,22 +115,27 @@ class GcsDiffFileLister {
     // (extracted from the filename) to its asynchronously-loaded metadata, keeping only files with
     // an upper checkpoint time > fromTime.
     TreeMap<DateTime, ListenableFuture<BlobInfo>> upperBoundTimesToBlobInfo = new TreeMap<>();
-    ImmutableList<String> strippedFilenames;
+    String commitLogDiffPrefix = getCommitLogDiffPrefix(fromTime, toTime);
+    ImmutableList<String> filenames;
     try {
-      strippedFilenames = gcsUtils.listFolderObjects(gcsBucket, DIFF_FILE_PREFIX);
+      filenames =
+          gcsUtils.listFolderObjects(gcsBucket, commitLogDiffPrefix).stream()
+              .map(s -> commitLogDiffPrefix + s)
+              .collect(toImmutableList());
     } catch (IOException e) {
       throw new RuntimeException(e);
     }
     DateTime lastUpperBoundTime = START_OF_TIME;
 
     TreeMap<DateTime, BlobInfo> sequence = new TreeMap<>();
+    ListeningExecutorService executor = executorProvider.get();
     try {
-      for (String strippedFilename : strippedFilenames) {
-        final String filename = DIFF_FILE_PREFIX + strippedFilename;
+      for (String filename : filenames) {
+        String strippedFilename = filename.replaceFirst(DIFF_FILE_PREFIX, "");
         DateTime upperBoundTime = DateTime.parse(strippedFilename);
         if (isInRange(upperBoundTime, fromTime, toTime)) {
           upperBoundTimesToBlobInfo.put(
-              upperBoundTime, lazyExecutor.get().submit(() -> getBlobInfo(gcsBucket, filename)));
+              upperBoundTime, executor.submit(() -> getBlobInfo(gcsBucket, filename)));
           lastUpperBoundTime = latestOf(upperBoundTime, lastUpperBoundTime);
         }
       }
@@ -173,7 +180,7 @@ class GcsDiffFileLister {
           "Unable to compute commit diff history, there are either gaps or forks in the history "
               + "file set.  Check log for details.");
     } finally {
-      lazyExecutor.get().shutdown();
+      executor.shutdown();
     }
 
     logger.atInfo().log(
@@ -198,4 +205,36 @@ class GcsDiffFileLister {
   private BlobInfo getBlobInfo(String gcsBucket, String filename) {
     return gcsUtils.getBlobInfo(BlobId.of(gcsBucket, filename));
   }
+
+  /**
+   * Returns a prefix guaranteed to cover all commit log diff files in the given range.
+   *
+   * <p>The listObjects call can be fairly slow if we search over many thousands or tens of
+   * thousands of files, so we restrict the search space. The commit logs have a file format of
+   * "commit_diff_until_2021-05-11T06:48:00.070Z" so we can often filter down as far as the hour.
+   *
+   * <p>Here, we get the longest prefix possible based on which fields (year, month, day, hour) the
+   * times in question have in common.
+   */
+  @VisibleForTesting
+  static String getCommitLogDiffPrefix(DateTime from, @Nullable DateTime to) {
+    StringBuilder result = new StringBuilder(DIFF_FILE_PREFIX);
+    if (to == null || from.getYear() != to.getYear()) {
+      return result.toString();
+    }
+    result.append(from.getYear()).append('-');
+    if (from.getMonthOfYear() != to.getMonthOfYear()) {
+      return result.toString();
+    }
+    result.append(String.format("%02d-", from.getMonthOfYear()));
+    if (from.getDayOfMonth() != to.getDayOfMonth()) {
+      return result.toString();
+    }
+    result.append(String.format("%02dT", from.getDayOfMonth()));
+    if (from.getHourOfDay() != to.getHourOfDay()) {
+      return result.toString();
+    }
+    result.append(String.format("%02d:", from.getHourOfDay()));
+    return result.toString();
+  }
 }

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

@@ -20,13 +20,15 @@ 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;
+import static google.registry.util.DateTimeUtils.isAtOrAfter;
+import static google.registry.util.DateTimeUtils.isBeforeOrAt;
 import static javax.servlet.http.HttpServletResponse.SC_NO_CONTENT;
+import static javax.servlet.http.HttpServletResponse.SC_OK;
 import static org.joda.time.Duration.standardHours;
 
 import com.google.appengine.api.datastore.Entity;
 import com.google.appengine.api.datastore.Key;
 import com.google.cloud.storage.BlobInfo;
-import com.google.common.base.Joiner;
 import com.google.common.collect.ImmutableList;
 import com.google.common.flogger.FluentLogger;
 import google.registry.config.RegistryConfig.Config;
@@ -120,26 +122,15 @@ public class ReplayCommitLogsToSqlAction implements Runnable {
     }
     try {
       logger.atInfo().log("Beginning replay of commit logs.");
-      ImmutableList<BlobInfo> commitLogFiles = getFilesToReplay();
+      String resultMessage;
       if (dryRun) {
-        response.setStatus(HttpServletResponse.SC_OK);
-        ImmutableList<String> filenames =
-            commitLogFiles.stream()
-                .limit(10)
-                .map(file -> file.getName())
-                .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);
+        resultMessage = executeDryRun();
       } else {
-        replayFiles(commitLogFiles);
-        response.setStatus(HttpServletResponse.SC_OK);
-        String message = "ReplayCommitLogsToSqlAction completed successfully.";
-        response.setPayload(message);
-        logger.atInfo().log(message);
+        resultMessage = replayFiles();
       }
+      response.setStatus(SC_OK);
+      response.setPayload(resultMessage);
+      logger.atInfo().log(resultMessage);
     } catch (Throwable t) {
       String message = "Errored out replaying files.";
       logger.atSevere().withCause(t).log(message);
@@ -150,52 +141,77 @@ public class ReplayCommitLogsToSqlAction implements Runnable {
     }
   }
 
-  private ImmutableList<BlobInfo> getFilesToReplay() {
+  private String executeDryRun() {
     // 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<BlobInfo> commitLogFiles =
-        diffLister.listDiffFiles(gcsBucket, fromTime, /* current time */ null);
-    logger.atInfo().log("Found %d new commit log files to process.", commitLogFiles.size());
-    return commitLogFiles;
+    DateTime searchStartTime = jpaTm().transact(() -> SqlReplayCheckpoint.get().plusMillis(1));
+    // Search through the end of the hour
+    DateTime searchEndTime =
+        searchStartTime.withMinuteOfHour(59).withSecondOfMinute(59).withMillisOfSecond(999);
+    ImmutableList<String> fileBatch =
+        diffLister.listDiffFiles(gcsBucket, searchStartTime, searchEndTime).stream()
+            .map(BlobInfo::getName)
+            .collect(toImmutableList());
+    return String.format(
+        "Running in dry-run mode, the first set of commit log files processed would be from "
+            + "searching from %s to %s and would contain %d file(s). They are (limit 10): \n%s",
+        searchStartTime,
+        searchEndTime,
+        fileBatch.size(),
+        fileBatch.stream().limit(10).collect(toImmutableList()));
   }
 
-  private void replayFiles(ImmutableList<BlobInfo> commitLogFiles) {
+  private String replayFiles() {
     DateTime replayTimeoutTime = clock.nowUtc().plus(REPLAY_TIMEOUT_DURATION);
-    int processedFiles = 0;
-    for (BlobInfo 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;
+    DateTime searchStartTime = jpaTm().transact(() -> SqlReplayCheckpoint.get().plusMillis(1));
+    int filesProcessed = 0;
+    // Starting from one millisecond after the last file we processed, search for and import files
+    // one hour at a time until we catch up to the current time or we hit the replay timeout (in
+    // which case the next run will pick up from where we leave off).
+    //
+    // We use hour-long batches because GCS supports filename prefix-based searches.
+    while (true) {
+      if (isAtOrAfter(clock.nowUtc(), replayTimeoutTime)) {
+        return String.format(
+            "Reached max execution time after replaying %d file(s).", filesProcessed);
       }
+      if (isBeforeOrAt(clock.nowUtc(), searchStartTime)) {
+        return String.format(
+            "Caught up to current time after replaying %d file(s).", filesProcessed);
+      }
+      // Search through the end of the hour
+      DateTime searchEndTime =
+          searchStartTime.withMinuteOfHour(59).withSecondOfMinute(59).withMillisOfSecond(999);
+      ImmutableList<BlobInfo> fileBatch =
+          diffLister.listDiffFiles(gcsBucket, searchStartTime, searchEndTime);
+      if (fileBatch.isEmpty()) {
+        logger.atInfo().log(
+            "No remaining files found in hour %s, continuing search in the next hour.",
+            searchStartTime.toString("yyyy-MM-dd HH"));
+      }
+      for (BlobInfo file : fileBatch) {
+        jpaTm().transact(() -> processFile(file));
+        filesProcessed++;
+        if (clock.nowUtc().isAfter(replayTimeoutTime)) {
+          return String.format(
+              "Reached max execution time after replaying %d file(s).", filesProcessed);
+        }
+      }
+      searchStartTime = searchEndTime.plusMillis(1);
     }
-    logger.atInfo().log("Replayed %d commit log files to SQL successfully.", processedFiles);
   }
 
   private void processFile(BlobInfo metadata) {
-    logger.atInfo().log(
-        "Processing commit log file %s of size %d B.", metadata.getName(), metadata.getSize());
     try (InputStream input = gcsUtils.openInputStream(metadata.getBlobId())) {
       // 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.getName());
       allTransactions.forEach(this::replayTransaction);
       // if we succeeded, set the last-seen time
       DateTime checkpoint = DateTime.parse(metadata.getName().substring(DIFF_FILE_PREFIX.length()));
       SqlReplayCheckpoint.set(checkpoint);
       logger.atInfo().log(
-          "Replayed %d transactions from commit log file %s.",
-          allTransactions.size(), metadata.getName());
+          "Replayed %d transactions from commit log file %s with size %d B.",
+          allTransactions.size(), metadata.getName(), metadata.getSize());
     } catch (IOException e) {
       throw new RuntimeException(
           "Errored out while replaying commit log file " + metadata.getName(), e);
@@ -223,7 +239,7 @@ public class ReplayCommitLogsToSqlAction implements Runnable {
           .toSqlEntity()
           .ifPresent(
               sqlEntity -> {
-                ReplaySpecializer.beforeSqlSave(sqlEntity);
+                sqlEntity.beforeSqlSaveOnReplay();
                 jpaTm().put(sqlEntity);
               });
     } else {

core/src/main/java/google/registry/beam/initsql/CloudSqlCredentialDecryptor.java

@@ -1,50 +0,0 @@
-// Copyright 2020 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.initsql;
-
-import static com.google.common.base.Preconditions.checkArgument;
-
-import com.google.api.services.cloudkms.v1.model.DecryptRequest;
-import com.google.common.base.Strings;
-import google.registry.config.RegistryConfig.Config;
-import google.registry.keyring.kms.KmsConnection;
-import java.nio.charset.StandardCharsets;
-import java.util.Base64;
-import javax.inject.Inject;
-
-/**
- * Decrypts data using Cloud KMS, with the same crypto key with which Cloud SQL credential files on
- * GCS was encrypted. See {@link BackupPaths#getCloudSQLCredentialFilePatterns} for more
- * information.
- */
-public class CloudSqlCredentialDecryptor {
-
-  private static final String CRYPTO_KEY_NAME = "nomulus-tool-key";
-  private final KmsConnection kmsConnection;
-
-  @Inject
-  CloudSqlCredentialDecryptor(@Config("beamKmsConnection") KmsConnection kmsConnection) {
-    this.kmsConnection = kmsConnection;
-  }
-
-  public String decrypt(String data) {
-    checkArgument(!Strings.isNullOrEmpty(data), "Null or empty data.");
-    byte[] ciphertext = Base64.getDecoder().decode(data);
-    // Re-encode for Cloud KMS JSON REST API, invoked through kmsConnection.
-    String urlSafeCipherText = new DecryptRequest().encodeCiphertext(ciphertext).getCiphertext();
-    return new String(
-        kmsConnection.decrypt(CRYPTO_KEY_NAME, urlSafeCipherText), StandardCharsets.UTF_8);
-  }
-}

core/src/main/java/google/registry/beam/initsql/SqlAccessInfo.java

@@ -1,45 +0,0 @@
-// Copyright 2020 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.initsql;
-
-import com.google.auto.value.AutoValue;
-import java.util.Optional;
-
-/**
- * Information needed to connect to a database, including JDBC URL, user name, password, and in the
- * case of Cloud SQL, the database instance's name.
- */
-@AutoValue
-abstract class SqlAccessInfo {
-
-  abstract String jdbcUrl();
-
-  abstract String user();
-
-  abstract String password();
-
-  abstract Optional<String> cloudSqlInstanceName();
-
-  public static SqlAccessInfo createCloudSqlAccessInfo(
-      String sqlInstanceName, String username, String password) {
-    return new AutoValue_SqlAccessInfo(
-        "jdbc:postgresql://google/postgres", username, password, Optional.of(sqlInstanceName));
-  }
-
-  public static SqlAccessInfo createLocalSqlAccessInfo(
-      String jdbcUrl, String username, String password) {
-    return new AutoValue_SqlAccessInfo(jdbcUrl, username, password, Optional.empty());
-  }
-}

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

@@ -1267,16 +1267,34 @@ public final class RegistryConfig {
 
     @Provides
     @Config("expirationWarningDays")
-    public static int provideDaysToExpiration(RegistryConfigSettings config) {
+    public static int provideExpirationWarningDays(RegistryConfigSettings config) {
       return config.sslCertificateValidation.expirationWarningDays;
     }
 
+    @Provides
+    @Config("expirationWarningIntervalDays")
+    public static int provideExpirationWarningIntervalDays(RegistryConfigSettings config) {
+      return config.sslCertificateValidation.expirationWarningIntervalDays;
+    }
+
     @Provides
     @Config("minimumRsaKeyLength")
     public static int provideMinimumRsaKeyLength(RegistryConfigSettings config) {
       return config.sslCertificateValidation.minimumRsaKeyLength;
     }
 
+    @Provides
+    @Config("expirationWarningEmailBodyText")
+    public static String provideExpirationWarningEmailBodyText(RegistryConfigSettings config) {
+      return config.sslCertificateValidation.expirationWarningEmailBodyText;
+    }
+
+    @Provides
+    @Config("expirationWarningEmailSubjectText")
+    public static String provideExpirationWarningEmailSubjectText(RegistryConfigSettings config) {
+      return config.sslCertificateValidation.expirationWarningEmailSubjectText;
+    }
+
     @Provides
     @Config("allowedEcdsaCurves")
     public static ImmutableSet<String> provideAllowedEcdsaCurves(RegistryConfigSettings config) {
@@ -1476,21 +1494,6 @@ public final class RegistryConfig {
     return CONFIG_SETTINGS.get().hibernate.hikariIdleTimeout;
   }
 
-  /**
-   * Returns whether to replicate cloud SQL transactions to datastore.
-   *
-   * <p>If true, all cloud SQL transactions will be persisted as TransactionEntity objects in the
-   * Transaction table and replayed against datastore in a cron job.
-   */
-  public static boolean getCloudSqlReplicateTransactions() {
-    return CONFIG_SETTINGS.get().cloudSql.replicateTransactions;
-  }
-
-  @VisibleForTesting
-  public static void overrideCloudSqlReplicateTransactions(boolean replicateTransactions) {
-    CONFIG_SETTINGS.get().cloudSql.replicateTransactions = replicateTransactions;
-  }
-
   /** Returns the roid suffix to be used for the roids of all contacts and hosts. */
   public static String getContactAndHostRoidSuffix() {
     return CONFIG_SETTINGS.get().registryPolicy.contactAndHostRoidSuffix;

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

@@ -126,7 +126,6 @@ public class RegistryConfigSettings {
     // TODO(05012021): remove username field after it is removed from all yaml files.
     public String username;
     public String instanceConnectionName;
-    public boolean replicateTransactions;
   }
 
   /** Configuration for Apache Beam (Cloud Dataflow). */
@@ -228,7 +227,10 @@ public class RegistryConfigSettings {
   public static class SslCertificateValidation {
     public Map<String, Integer> maxValidityDaysSchedule;
     public int expirationWarningDays;
+    public int expirationWarningIntervalDays;
     public int minimumRsaKeyLength;
     public Set<String> allowedEcdsaCurves;
+    public String expirationWarningEmailBodyText;
+    public String expirationWarningEmailSubjectText;
   }
 }

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

@@ -227,9 +227,6 @@ cloudSql:
   jdbcUrl: jdbc:postgresql://localhost
   # This name is used by Cloud SQL when connecting to the database.
   instanceConnectionName: project-id:region:instance-id
-  # Set this to true to replicate cloud SQL transactions to datastore in the
-  # background.
-  replicateTransactions: false
 
 cloudDns:
   # Set both properties to null in Production.
@@ -452,6 +449,15 @@ sslCertificateValidation:
   # The number of days before a certificate expires that indicates the
   # certificate is nearing expiration and warnings should be sent.
   expirationWarningDays: 30
+  # The minimum number of days between two successive expiring notification emails.
+  expirationWarningIntervalDays: 15
+  # Text for expiring certificate notification email subject.
+  expirationWarningEmailSubjectText: Certificate Expring Within 30 Days.
+  # Text for expiring certificate notification email body that accepts 3 parameters:
+  # registrar name, certificate type, and expiration date, respectively.
+  expirationWarningEmailBodyText: |
+    Hello Registrar %s,
+       The %s certificate is expiring on %s.
   # The minimum number of bits an RSA key must contain.
   minimumRsaKeyLength: 2048
   # The ECDSA curves that are allowed for public keys.

core/src/main/java/google/registry/flows/certs/CertificateChecker.java

@@ -24,6 +24,7 @@ import google.registry.config.RegistryConfig.Config;
 import google.registry.util.Clock;
 import google.registry.util.DateTimeUtils;
 import java.io.ByteArrayInputStream;
+import java.io.StringWriter;
 import java.security.PublicKey;
 import java.security.cert.CertificateException;
 import java.security.cert.CertificateFactory;
@@ -32,10 +33,15 @@ import java.security.interfaces.ECPublicKey;
 import java.security.interfaces.RSAPublicKey;
 import java.util.Date;
 import java.util.stream.Collectors;
+import javax.annotation.Nullable;
 import javax.inject.Inject;
 import org.bouncycastle.jcajce.provider.asymmetric.util.EC5Util;
 import org.bouncycastle.jce.ECNamedCurveTable;
 import org.bouncycastle.jce.spec.ECNamedCurveParameterSpec;
+import org.bouncycastle.jce.spec.ECParameterSpec;
+import org.bouncycastle.openssl.jcajce.JcaMiscPEMGenerator;
+import org.bouncycastle.util.io.pem.PemObjectGenerator;
+import org.bouncycastle.util.io.pem.PemWriter;
 import org.joda.time.DateTime;
 import org.joda.time.Days;
 
@@ -43,10 +49,11 @@ import org.joda.time.Days;
 public class CertificateChecker {
 
   private final ImmutableSortedMap<DateTime, Integer> maxValidityLengthSchedule;
-  private final int daysToExpiration;
+  private final int expirationWarningDays;
   private final int minimumRsaKeyLength;
   private final Clock clock;
   private final ImmutableSet<String> allowedEcdsaCurves;
+  private final int expirationWarningIntervalDays;
 
   /**
    * Constructs a CertificateChecker instance with the specified configuration parameters.
@@ -72,6 +79,7 @@ public class CertificateChecker {
       @Config("maxValidityDaysSchedule")
           ImmutableSortedMap<DateTime, Integer> maxValidityDaysSchedule,
       @Config("expirationWarningDays") int expirationWarningDays,
+      @Config("expirationWarningIntervalDays") int expirationWarningIntervalDays,
       @Config("minimumRsaKeyLength") int minimumRsaKeyLength,
       @Config("allowedEcdsaCurves") ImmutableSet<String> allowedEcdsaCurves,
       Clock clock) {
@@ -79,12 +87,46 @@ public class CertificateChecker {
         maxValidityDaysSchedule.containsKey(START_OF_TIME),
         "Max validity length schedule must contain an entry for START_OF_TIME");
     this.maxValidityLengthSchedule = maxValidityDaysSchedule;
-    this.daysToExpiration = expirationWarningDays;
+    this.expirationWarningDays = expirationWarningDays;
     this.minimumRsaKeyLength = minimumRsaKeyLength;
     this.allowedEcdsaCurves = allowedEcdsaCurves;
+    this.expirationWarningIntervalDays = expirationWarningIntervalDays;
     this.clock = clock;
   }
 
+  private static int getValidityLengthInDays(X509Certificate certificate) {
+    DateTime start = DateTime.parse(certificate.getNotBefore().toInstant().toString());
+    DateTime end = DateTime.parse(certificate.getNotAfter().toInstant().toString());
+    return Days.daysBetween(start.withTimeAtStartOfDay(), end.withTimeAtStartOfDay()).getDays();
+  }
+
+  /** Checks if the curve used for a public key is in the list of acceptable curves. */
+  private static boolean checkCurveName(PublicKey key, ImmutableSet<String> allowedEcdsaCurves) {
+    ECParameterSpec params;
+    // These 2 different instances of PublicKey need to be handled separately since their OIDs are
+    // encoded differently. More details on this can be found at
+    // https://stackoverflow.com/questions/49895713/how-to-find-the-matching-curve-name-from-an-ecpublickey.
+    if (key instanceof ECPublicKey) {
+      ECPublicKey ecKey = (ECPublicKey) key;
+      params = EC5Util.convertSpec(ecKey.getParams(), false);
+    } else if (key instanceof org.bouncycastle.jce.interfaces.ECPublicKey) {
+      org.bouncycastle.jce.interfaces.ECPublicKey ecKey =
+          (org.bouncycastle.jce.interfaces.ECPublicKey) key;
+      params = ecKey.getParameters();
+    } else {
+      throw new IllegalArgumentException("Unrecognized instance of PublicKey.");
+    }
+    return allowedEcdsaCurves.stream()
+        .anyMatch(
+            curve -> {
+              ECNamedCurveParameterSpec cParams = ECNamedCurveTable.getParameterSpec(curve);
+              return cParams.getN().equals(params.getN())
+                  && cParams.getH().equals(params.getH())
+                  && cParams.getCurve().equals(params.getCurve())
+                  && cParams.getG().equals(params.getG());
+            });
+  }
+
   /**
    * Checks the given certificate string for violations and throws an exception if any violations
    * exist.
@@ -156,67 +198,57 @@ public class CertificateChecker {
    * the violations the certificate has.
    */
   public ImmutableSet<CertificateViolation> checkCertificate(String certificateString) {
-    X509Certificate certificate;
+    return checkCertificate(getCertificate(certificateString));
+  }
 
+  /** Converts the given string to a certificate object. */
+  public X509Certificate getCertificate(String certificateStr) {
+    X509Certificate certificate;
     try {
       certificate =
           (X509Certificate)
               CertificateFactory.getInstance("X509")
-                  .generateCertificate(new ByteArrayInputStream(certificateString.getBytes(UTF_8)));
+                  .generateCertificate(new ByteArrayInputStream(certificateStr.getBytes(UTF_8)));
     } catch (CertificateException e) {
-      throw new IllegalArgumentException("Unable to read given certificate.");
+      throw new IllegalArgumentException(
+          String.format("Unable to read given certificate %s", certificateStr), e);
     }
-
-    return checkCertificate(certificate);
+    return certificate;
   }
 
-  /**
-   * Returns whether the certificate is nearing expiration.
-   *
-   * <p>Note that this is <i>all</i> that it checks. The certificate itself may well be expired or
-   * not yet valid and this message will still return false. So you definitely want to pair a call
-   * to this method with a call to {@link #checkCertificate} to determine other issues with the
-   * certificate that may be occurring.
-   */
-  public boolean isNearingExpiration(X509Certificate certificate) {
-    Date nearingExpirationDate =
-        DateTime.parse(certificate.getNotAfter().toInstant().toString())
-            .minusDays(daysToExpiration)
-            .toDate();
-    return clock.nowUtc().toDate().after(nearingExpirationDate);
-  }
-
-  private static int getValidityLengthInDays(X509Certificate certificate) {
-    DateTime start = DateTime.parse(certificate.getNotBefore().toInstant().toString());
-    DateTime end = DateTime.parse(certificate.getNotAfter().toInstant().toString());
-    return Days.daysBetween(start.withTimeAtStartOfDay(), end.withTimeAtStartOfDay()).getDays();
-  }
-
-  /** Checks if the curve used for a public key is in the list of acceptable curves. */
-  private static boolean checkCurveName(PublicKey key, ImmutableSet<String> allowedEcdsaCurves) {
-    org.bouncycastle.jce.spec.ECParameterSpec params;
-    // These 2 different instances of PublicKey need to be handled separately since their OIDs are
-    // encoded differently. More details on this can be found at
-    // https://stackoverflow.com/questions/49895713/how-to-find-the-matching-curve-name-from-an-ecpublickey.
-    if (key instanceof ECPublicKey) {
-      ECPublicKey ecKey = (ECPublicKey) key;
-      params = EC5Util.convertSpec(ecKey.getParams(), false);
-    } else if (key instanceof org.bouncycastle.jce.interfaces.ECPublicKey) {
-      org.bouncycastle.jce.interfaces.ECPublicKey ecKey =
-          (org.bouncycastle.jce.interfaces.ECPublicKey) key;
-      params = ecKey.getParameters();
-    } else {
-      throw new IllegalArgumentException("Unrecognized instance of PublicKey.");
+  /** Serializes the certificate object to a certificate string. */
+  public String serializeCertificate(X509Certificate certificate) throws Exception {
+    StringWriter sw = new StringWriter();
+    try (PemWriter pw = new PemWriter(sw)) {
+      PemObjectGenerator generator = new JcaMiscPEMGenerator(certificate);
+      pw.writeObject(generator);
     }
-    return allowedEcdsaCurves.stream()
-        .anyMatch(
-            curve -> {
-              ECNamedCurveParameterSpec cParams = ECNamedCurveTable.getParameterSpec(curve);
-              return cParams.getN().equals(params.getN())
-                  && cParams.getH().equals(params.getH())
-                  && cParams.getCurve().equals(params.getCurve())
-                  && cParams.getG().equals(params.getG());
-            });
+    return sw.toString();
+  }
+
+  /** Returns whether the client should receive a notification email. */
+  public boolean shouldReceiveExpiringNotification(
+      @Nullable DateTime lastExpiringNotificationSentDate, String certificateStr) {
+    X509Certificate certificate = getCertificate(certificateStr);
+    DateTime now = clock.nowUtc();
+    // expiration date is one day after lastValidDate
+    Date lastValidDate = certificate.getNotAfter();
+    if (lastValidDate.before(now.toDate())) {
+      return false;
+    }
+    /*
+     * Client should receive a notification if :
+     *    1) client has never received notification and the certificate has entered
+     *    the expiring period, OR
+     *    2) client has received notification but the interval between now and
+     *    lastExpiringNotificationSentDate is greater than expirationWarningIntervalDays.
+     */
+    return !lastValidDate.after(now.plusDays(expirationWarningDays).toDate())
+        && (lastExpiringNotificationSentDate == null
+            || !lastExpiringNotificationSentDate
+                .plusDays(expirationWarningIntervalDays)
+                .toDate()
+                .after(now.toDate()));
   }
 
   private String getViolationDisplayMessage(CertificateViolation certificateViolation) {

core/src/main/java/google/registry/keyring/kms/KmsKeyring.java

@@ -28,14 +28,8 @@ import org.bouncycastle.openpgp.PGPKeyPair;
 import org.bouncycastle.openpgp.PGPPrivateKey;
 import org.bouncycastle.openpgp.PGPPublicKey;
 
-/**
- * A {@link Keyring} implementation which stores encrypted secrets in Datastore and decrypts them
- * using encryption keys stored in Cloud KMS.
- *
- * @see <a href="https://cloud.google.com/kms/docs/">Google Cloud Key Management Service
- *     Documentation</a>
- */
-// TODO(2021-07-01): rename this class to SecretManagerKeyring and delete KmsSecretRevision
+/** A {@link Keyring} implementation which stores sensitive data in the Secret Manager. */
+// TODO(2021-08-01): rename this class to SecretManagerKeyring and update config files.
 public class KmsKeyring implements Keyring {
 
   /** Key labels for private key secrets. */

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

@@ -43,8 +43,6 @@ import google.registry.model.registry.Registry;
 import google.registry.model.registry.label.PremiumList;
 import google.registry.model.registry.label.ReservedList;
 import google.registry.model.reporting.HistoryEntry;
-import google.registry.model.server.KmsSecret;
-import google.registry.model.server.KmsSecretRevision;
 import google.registry.model.server.Lock;
 import google.registry.model.server.ServerSecret;
 import google.registry.model.tmch.ClaimsList;
@@ -88,8 +86,6 @@ public final class EntityClasses {
           HistoryEntry.class,
           HostHistory.class,
           HostResource.class,
-          KmsSecret.class,
-          KmsSecretRevision.class,
           LastSqlTransaction.class,
           Lock.class,
           PollMessage.class,

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

@@ -225,7 +225,7 @@ public class DatabaseMigrationStateSchedule extends CrossTldSingleton
   @VisibleForTesting
   static TimedTransitionProperty<MigrationState, MigrationStateTransition> getUncached() {
     return ofyTm()
-        .transact(
+        .transactNew(
             () ->
                 ofyTm()
                     .loadSingleton(DatabaseMigrationStateSchedule.class)

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

@@ -136,9 +136,11 @@ public class ContactHistory extends HistoryEntry implements SqlEntity {
   }
 
   // Used to fill out the contactBase field during asynchronous replay
-  public static void beforeSqlSave(ContactHistory contactHistory) {
-    contactHistory.contactBase =
-        jpaTm().loadByKey(VKey.createSql(ContactResource.class, contactHistory.getContactRepoId()));
+  @Override
+  public void beforeSqlSaveOnReplay() {
+    if (contactBase == null) {
+      contactBase = jpaTm().getEntityManager().find(ContactResource.class, getContactRepoId());
+    }
   }
 
   /** Class to represent the composite primary key of {@link ContactHistory} entity. */

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

@@ -294,9 +294,26 @@ public class DomainHistory extends HistoryEntry implements SqlEntity {
   }
 
   // Used to fill out the domainContent field during asynchronous replay
-  public static void beforeSqlSave(DomainHistory domainHistory) {
-    domainHistory.domainContent =
-        jpaTm().loadByKey(VKey.createSql(DomainBase.class, domainHistory.getDomainRepoId()));
+  @Override
+  public void beforeSqlSaveOnReplay() {
+    if (domainContent == null) {
+      domainContent = jpaTm().getEntityManager().find(DomainBase.class, getDomainRepoId());
+      fillAuxiliaryFieldsFromDomain(this);
+    }
+  }
+
+  private static void fillAuxiliaryFieldsFromDomain(DomainHistory domainHistory) {
+    if (domainHistory.domainContent != null) {
+      domainHistory.nsHosts = nullToEmptyImmutableCopy(domainHistory.domainContent.nsHosts);
+      domainHistory.dsDataHistories =
+          nullToEmptyImmutableCopy(domainHistory.domainContent.getDsData()).stream()
+              .map(dsData -> DomainDsDataHistory.createFrom(domainHistory.id, dsData))
+              .collect(toImmutableSet());
+      domainHistory.gracePeriodHistories =
+          nullToEmptyImmutableCopy(domainHistory.domainContent.getGracePeriods()).stream()
+              .map(gracePeriod -> GracePeriodHistory.createFrom(domainHistory.id, gracePeriod))
+              .collect(toImmutableSet());
+    }
   }
 
   /** Class to represent the composite primary key of {@link DomainHistory} entity. */
@@ -391,17 +408,7 @@ public class DomainHistory extends HistoryEntry implements SqlEntity {
       // Note that we cannot assert that instance.domainContent is not null here because this
       // builder is also used to convert legacy HistoryEntry objects to DomainHistory, when
       // domainContent is not available.
-      if (instance.domainContent != null) {
-        instance.nsHosts = nullToEmptyImmutableCopy(instance.domainContent.nsHosts);
-        instance.dsDataHistories =
-            nullToEmptyImmutableCopy(instance.domainContent.getDsData()).stream()
-                .map(dsData -> DomainDsDataHistory.createFrom(instance.id, dsData))
-                .collect(toImmutableSet());
-        instance.gracePeriodHistories =
-            nullToEmptyImmutableCopy(instance.domainContent.getGracePeriods()).stream()
-                .map(gracePeriod -> GracePeriodHistory.createFrom(instance.id, gracePeriod))
-                .collect(toImmutableSet());
-      }
+      fillAuxiliaryFieldsFromDomain(instance);
       return instance;
     }
   }

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

@@ -136,9 +136,11 @@ public class HostHistory extends HistoryEntry implements SqlEntity {
   }
 
   // Used to fill out the hostBase field during asynchronous replay
-  public static void beforeSqlSave(HostHistory hostHistory) {
-    hostHistory.hostBase =
-        jpaTm().loadByKey(VKey.createSql(HostResource.class, hostHistory.getHostRepoId()));
+  @Override
+  public void beforeSqlSaveOnReplay() {
+    if (hostBase == null) {
+      hostBase = jpaTm().getEntityManager().find(HostResource.class, getHostRepoId());
+    }
   }
 
   /** Class to represent the composite primary key of {@link HostHistory} entity. */

core/src/main/java/google/registry/model/registrar/Registrar.java

@@ -37,6 +37,7 @@ import static google.registry.persistence.transaction.TransactionManagerFactory.
 import static google.registry.persistence.transaction.TransactionManagerUtil.transactIfJpaTm;
 import static google.registry.util.CollectionUtils.nullToEmptyImmutableCopy;
 import static google.registry.util.CollectionUtils.nullToEmptyImmutableSortedCopy;
+import static google.registry.util.DateTimeUtils.START_OF_TIME;
 import static google.registry.util.PasswordUtils.SALT_SUPPLIER;
 import static google.registry.util.PasswordUtils.hashPassword;
 import static google.registry.util.PreconditionsUtils.checkArgumentNotNull;
@@ -458,6 +459,14 @@ public class Registrar extends ImmutableObject
   /** The time that the certificate was last updated. */
   DateTime lastCertificateUpdateTime;
 
+  /** The time that an expiring certificate notification email was sent to the registrar. */
+  DateTime lastExpiringCertNotificationSentDate = START_OF_TIME;
+
+  /**
+   * The time that an expiring failover certificate notification email was sent to the registrar.
+   */
+  DateTime lastExpiringFailoverCertNotificationSentDate = START_OF_TIME;
+
   /** Telephone support passcode (5-digit numeric) */
   String phonePasscode;
 
@@ -508,6 +517,14 @@ public class Registrar extends ImmutableObject
     return lastCertificateUpdateTime;
   }
 
+  public DateTime getLastExpiringCertNotificationSentDate() {
+    return lastExpiringCertNotificationSentDate;
+  }
+
+  public DateTime getLastExpiringFailoverCertNotificationSentDate() {
+    return lastExpiringFailoverCertNotificationSentDate;
+  }
+
   public String getRegistrarName() {
     return registrarName;
   }
@@ -671,6 +688,10 @@ public class Registrar extends ImmutableObject
         .putString("creationTime", creationTime.getTimestamp())
         .putString("lastUpdateTime", lastUpdateTime.getTimestamp())
         .putString("lastCertificateUpdateTime", lastCertificateUpdateTime)
+        .putString("lastExpiringCertNotificationSentDate", lastExpiringCertNotificationSentDate)
+        .putString(
+            "lastExpiringFailoverCertNotificationSentDate",
+            lastExpiringFailoverCertNotificationSentDate)
         .put("registrarName", registrarName)
         .put("type", type)
         .put("state", state)
@@ -839,6 +860,19 @@ public class Registrar extends ImmutableObject
       return this;
     }
 
+    public Builder setLastExpiringCertNotificationSentDate(DateTime now) {
+      checkArgumentNotNull(now, "Registrar lastExpiringCertNotificationSentDate cannot be null");
+      getInstance().lastExpiringCertNotificationSentDate = now;
+      return this;
+    }
+
+    public Builder setLastExpiringFailoverCertNotificationSentDate(DateTime now) {
+      checkArgumentNotNull(
+          now, "Registrar lastExpiringFailoverCertNotificationSentDate cannot be null");
+      getInstance().lastExpiringFailoverCertNotificationSentDate = now;
+      return this;
+    }
+
     public Builder setFailoverClientCertificate(String clientCertificate, DateTime now) {
       clientCertificate = emptyToNull(clientCertificate);
       String clientCertificateHash = calculateHash(clientCertificate);

core/src/main/java/google/registry/model/server/KmsSecret.java

@@ -1,57 +0,0 @@
-// Copyright 2017 The Nomulus Authors. All Rights Reserved.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package google.registry.model.server;
-
-import static google.registry.model.common.EntityGroupRoot.getCrossTldKey;
-
-import com.googlecode.objectify.Key;
-import com.googlecode.objectify.annotation.Entity;
-import com.googlecode.objectify.annotation.Id;
-import com.googlecode.objectify.annotation.Parent;
-import google.registry.model.ImmutableObject;
-import google.registry.model.annotations.InCrossTld;
-import google.registry.model.annotations.ReportedOn;
-import google.registry.model.common.EntityGroupRoot;
-import google.registry.schema.replay.DatastoreOnlyEntity;
-
-/** Pointer to the latest {@link KmsSecretRevision}. */
-@Entity
-@ReportedOn
-@InCrossTld
-public class KmsSecret extends ImmutableObject implements DatastoreOnlyEntity {
-
-  /** The unique name of this {@link KmsSecret}. */
-  @Id String name;
-
-  @Parent Key<EntityGroupRoot> parent = getCrossTldKey();
-
-  /** The pointer to the latest {@link KmsSecretRevision}. */
-  Key<KmsSecretRevision> latestRevision;
-
-  public String getName() {
-    return name;
-  }
-
-  public Key<KmsSecretRevision> getLatestRevision() {
-    return latestRevision;
-  }
-
-  public static KmsSecret create(String name, KmsSecretRevision latestRevision) {
-    KmsSecret instance = new KmsSecret();
-    instance.name = name;
-    instance.latestRevision = Key.create(latestRevision);
-    return instance;
-  }
-}

core/src/main/java/google/registry/model/server/KmsSecretRevision.java

@@ -1,162 +0,0 @@
-// Copyright 2017 The Nomulus Authors. All Rights Reserved.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package google.registry.model.server;
-
-import static com.google.common.base.Preconditions.checkArgument;
-import static google.registry.model.common.EntityGroupRoot.getCrossTldKey;
-
-import com.googlecode.objectify.Key;
-import com.googlecode.objectify.annotation.Entity;
-import com.googlecode.objectify.annotation.Id;
-import com.googlecode.objectify.annotation.Ignore;
-import com.googlecode.objectify.annotation.OnLoad;
-import com.googlecode.objectify.annotation.Parent;
-import google.registry.model.Buildable;
-import google.registry.model.CreateAutoTimestamp;
-import google.registry.model.ImmutableObject;
-import google.registry.model.annotations.InCrossTld;
-import google.registry.model.annotations.ReportedOn;
-import google.registry.schema.replay.NonReplicatedEntity;
-import javax.persistence.Column;
-import javax.persistence.Index;
-import javax.persistence.PostLoad;
-import javax.persistence.Table;
-import javax.persistence.Transient;
-
-/**
- * An encrypted value.
- *
- * <p>Used to store passwords and other sensitive information in Datastore. Multiple versions of a
- * {@link KmsSecretRevision} may be persisted but only the latest version is primary. A key to the
- * primary version is stored by {@link KmsSecret#latestRevision}.
- *
- * <p>The value can be encrypted and decrypted using Cloud KMS.
- *
- * <p>Note that the primary key of this entity is {@link #revisionKey}, which is auto-generated by
- * the database. So, if a retry of insertion happens after the previous attempt unexpectedly
- * succeeds, we will end up with having two exact same revisions that differ only by revisionKey.
- * This is fine though, because we only use the revision with the highest revisionKey.
- *
- * <p>TODO(b/177567432): remove Datastore-specific fields post-Registry-3.0-migration and rename to
- * KmsSecret.
- *
- * @see <a href="https://cloud.google.com/kms/docs/">Google Cloud Key Management Service
- *     Documentation</a>
- * @see google.registry.keyring.kms.KmsKeyring
- */
-@Entity
-@ReportedOn
-@javax.persistence.Entity(name = "KmsSecret")
-@Table(indexes = {@Index(columnList = "secretName")})
-@InCrossTld
-public class KmsSecretRevision extends ImmutableObject implements NonReplicatedEntity {
-
-  /**
-   * The maximum allowable secret size. Although Datastore allows entities up to 1 MB in size,
-   * BigQuery imports of Datastore backups limit individual columns (entity attributes) to 64 KB.
-   */
-  private static final int MAX_SECRET_SIZE_BYTES = 64 * 1024 * 1024;
-
-  /**
-   * The revision of this secret.
-   *
-   * <p>TODO(b/177567432): change name of the variable to revisionId once we're off Datastore
-   */
-  @Id
-  @javax.persistence.Id
-  @Column(name = "revisionId")
-  long revisionKey;
-
-  /** The parent {@link KmsSecret} which contains metadata about this {@link KmsSecretRevision}. */
-  @Parent @Transient Key<KmsSecret> parent;
-  @Column(nullable = false)
-  @Ignore
-  String secretName;
-
-  /**
-   * The name of the {@code cryptoKeyVersion} associated with this {@link KmsSecretRevision}.
-   *
-   * <p>TODO: change name of the variable to cryptoKeyVersionName once we're off Datastore
-   *
-   * @see <a
-   *     href="https://cloud.google.com/kms/docs/reference/rest/v1/projects.locations.keyRings.cryptoKeys.cryptoKeyVersions">projects.locations.keyRings.cryptoKeys.cryptoKeyVersions</a>
-   */
-  @Column(nullable = false, name = "cryptoKeyVersionName")
-  String kmsCryptoKeyVersionName;
-
-  /**
-   * The base64-encoded encrypted value of this {@link KmsSecretRevision} as returned by the Cloud
-   * KMS API.
-   *
-   * @see <a
-   *     href="https://cloud.google.com/kms/docs/reference/rest/v1/projects.locations.keyRings.cryptoKeys/encrypt">projects.locations.keyRings.cryptoKeys.encrypt</a>
-   */
-  @Column(nullable = false)
-  String encryptedValue;
-
-  /** An automatically managed creation timestamp. */
-  @Column(nullable = false)
-  CreateAutoTimestamp creationTime = CreateAutoTimestamp.create(null);
-
-  public String getKmsCryptoKeyVersionName() {
-    return kmsCryptoKeyVersionName;
-  }
-
-  public String getEncryptedValue() {
-    return encryptedValue;
-  }
-
-  // When loading from SQL, fill out the Datastore-specific field
-  @PostLoad
-  void postLoad() {
-    parent = Key.create(getCrossTldKey(), KmsSecret.class, secretName);
-  }
-
-  // When loading from Datastore, fill out the SQL-specific field
-  @OnLoad
-  void onLoad() {
-    secretName = parent.getName();
-  }
-
-  /** A builder for constructing {@link KmsSecretRevision} entities, since they are immutable. */
-  public static class Builder extends Buildable.Builder<KmsSecretRevision> {
-
-    public Builder setKmsCryptoKeyVersionName(String kmsCryptoKeyVersionName) {
-      getInstance().kmsCryptoKeyVersionName = kmsCryptoKeyVersionName;
-      return this;
-    }
-
-    public Builder setEncryptedValue(String encryptedValue) {
-      checkArgument(
-          encryptedValue.length() <= MAX_SECRET_SIZE_BYTES,
-          "Secret is greater than %s bytes",
-          MAX_SECRET_SIZE_BYTES);
-
-      getInstance().encryptedValue = encryptedValue;
-      return this;
-    }
-
-    /**
-     * Set the parent {@link KmsSecret}.
-     *
-     * <p>The secret may not exist yet, so it is referred to by name rather than by reference.
-     */
-    public Builder setParent(String secretName) {
-      getInstance().parent = Key.create(getCrossTldKey(), KmsSecret.class, secretName);
-      getInstance().secretName = secretName;
-      return this;
-    }
-  }
-}

core/src/main/java/google/registry/model/server/KmsSecretRevisionSqlDao.java

@@ -1,53 +0,0 @@
-// Copyright 2020 The Nomulus Authors. All Rights Reserved.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package google.registry.model.server;
-
-import static com.google.common.base.Preconditions.checkArgument;
-import static com.google.common.base.Strings.isNullOrEmpty;
-import static google.registry.persistence.transaction.TransactionManagerFactory.jpaTm;
-import static google.registry.util.PreconditionsUtils.checkArgumentNotNull;
-
-import java.util.Optional;
-
-/**
- * A {@link KmsSecretRevision} DAO for Cloud SQL.
- *
- * <p>TODO(b/177567432): Rename this class to KmsSecretDao after migrating to Cloud SQL.
- */
-public class KmsSecretRevisionSqlDao {
-
-  private KmsSecretRevisionSqlDao() {}
-
-  /** Saves the given KMS secret revision. */
-  public static void save(KmsSecretRevision kmsSecretRevision) {
-    checkArgumentNotNull(kmsSecretRevision, "kmsSecretRevision cannot be null");
-    jpaTm().assertInTransaction();
-    jpaTm().put(kmsSecretRevision);
-  }
-
-  /** Returns the latest revision for the secret name given, or absent if nonexistent. */
-  public static Optional<KmsSecretRevision> getLatestRevision(String secretName) {
-    checkArgument(!isNullOrEmpty(secretName), "secretName cannot be null or empty");
-    jpaTm().assertInTransaction();
-    return jpaTm()
-        .query(
-            "FROM KmsSecret ks WHERE ks.revisionKey IN (SELECT MAX(revisionKey) FROM "
-                + "KmsSecret subKs WHERE subKs.secretName = :secretName)",
-            KmsSecretRevision.class)
-        .setParameter("secretName", secretName)
-        .getResultStream()
-        .findFirst();
-  }
-}

core/src/main/java/google/registry/persistence/transaction/JpaTransactionManagerImpl.java

@@ -30,14 +30,14 @@ import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.Streams;
 import com.google.common.flogger.FluentLogger;
-import google.registry.config.RegistryConfig;
 import google.registry.model.ImmutableObject;
+import google.registry.model.common.DatabaseMigrationStateSchedule;
+import google.registry.model.common.DatabaseMigrationStateSchedule.ReplayDirection;
 import google.registry.model.index.EppResourceIndex;
 import google.registry.model.index.ForeignKeyIndex.ForeignKeyContactIndex;
 import google.registry.model.index.ForeignKeyIndex.ForeignKeyDomainIndex;
 import google.registry.model.index.ForeignKeyIndex.ForeignKeyHostIndex;
 import google.registry.model.ofy.DatastoreTransactionManager;
-import google.registry.model.server.KmsSecret;
 import google.registry.model.tmch.ClaimsList.ClaimsListSingleton;
 import google.registry.persistence.JpaRetries;
 import google.registry.persistence.VKey;
@@ -94,8 +94,7 @@ public class JpaTransactionManagerImpl implements JpaTransactionManager {
           EppResourceIndex.class,
           ForeignKeyContactIndex.class,
           ForeignKeyDomainIndex.class,
-          ForeignKeyHostIndex.class,
-          KmsSecret.class);
+          ForeignKeyHostIndex.class);
 
   // EntityManagerFactory is thread safe.
   private final EntityManagerFactory emf;
@@ -105,6 +104,10 @@ public class JpaTransactionManagerImpl implements JpaTransactionManager {
   // synchronously.
   private final ThreadLocal<TransactionInfo> transactionInfo =
       ThreadLocal.withInitial(TransactionInfo::new);
+  // If this value is present, use it to determine whether or not to replay SQL transactions to
+  // Datastore, rather than using the schedule stored in Datastore.
+  private static ThreadLocal<Optional<Boolean>> replaySqlToDatastoreOverrideForTest =
+      ThreadLocal.withInitial(Optional::empty);
 
   public JpaTransactionManagerImpl(EntityManagerFactory emf, Clock clock) {
     this.emf = emf;
@@ -738,6 +741,16 @@ public class JpaTransactionManagerImpl implements JpaTransactionManager {
     return entity;
   }
 
+  /** Sets the override to always/never replay SQL transactions to Datastore. */
+  public static void setReplaySqlToDatastoreOverrideForTest(boolean replaySqlToDs) {
+    replaySqlToDatastoreOverrideForTest.set(Optional.of(replaySqlToDs));
+  }
+
+  /** Removes the replay-SQL-to-Datastore override; the migration schedule will then be used. */
+  public static void removeReplaySqlToDsOverrideForTest() {
+    replaySqlToDatastoreOverrideForTest.set(Optional.empty());
+  }
+
   private static class TransactionInfo {
     EntityManager entityManager;
     boolean inTransaction = false;
@@ -757,7 +770,14 @@ public class JpaTransactionManagerImpl implements JpaTransactionManager {
       checkArgumentNotNull(clock);
       inTransaction = true;
       transactionTime = clock.nowUtc();
-      if (withBackup && RegistryConfig.getCloudSqlReplicateTransactions()) {
+      if (withBackup
+          && replaySqlToDatastoreOverrideForTest
+              .get()
+              .orElseGet(
+                  () ->
+                      DatabaseMigrationStateSchedule.getValueAtTime(transactionTime)
+                          .getReplayDirection()
+                          .equals(ReplayDirection.SQL_TO_DATASTORE))) {
         contentsBuilder = new Transaction.Builder();
       }
     }

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

@@ -28,17 +28,11 @@ import java.lang.reflect.Method;
 public class ReplaySpecializer {
 
   public static void beforeSqlDelete(VKey<?> key) {
-    invokeMethod(key.getKind(), "beforeSqlDelete", key);
-  }
-
-  public static void beforeSqlSave(SqlEntity sqlEntity) {
-    invokeMethod(sqlEntity.getClass(), "beforeSqlSave", sqlEntity);
-  }
-
-  private static <T> void invokeMethod(Class<T> clazz, String methodName, Object argument) {
+    String methodName = "beforeSqlDelete";
+    Class<?> clazz = key.getKind();
     try {
-      Method method = clazz.getMethod(methodName, argument.getClass());
-      method.invoke(null, argument);
+      Method method = clazz.getMethod(methodName, VKey.class);
+      method.invoke(null, key);
     } catch (NoSuchMethodException e) {
       // Ignore, this just means that the class doesn't need this hook.
     } catch (IllegalAccessException e) {

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

@@ -26,4 +26,7 @@ import java.util.Optional;
 public interface SqlEntity {
 
   Optional<DatastoreEntity> toDatastoreEntity();
+
+  /** A method that will ber called before the object is saved to SQL in asynchronous replay. */
+  default void beforeSqlSaveOnReplay() {}
 }

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

@@ -53,7 +53,7 @@ import javax.inject.Inject;
  * However, since this is meant to be run during the Datastore-primary, SQL-secondary stage of the
  * migration, we want to make sure that we are using the most up-to-date version of the data. The
  * resource field of the history objects will be populated during asynchronous migration, e.g. in
- * {@link DomainHistory#beforeSqlSave(DomainHistory)}.
+ * {@link DomainHistory#beforeSqlSaveOnReplay}.
  */
 @Action(
     service = Action.Service.BACKEND,

core/src/main/resources/META-INF/persistence.xml

@@ -65,7 +65,6 @@
     <class>google.registry.model.registry.Registry</class>
     <class>google.registry.model.reporting.DomainTransactionRecord</class>
     <class>google.registry.model.reporting.Spec11ThreatMatch</class>
-    <class>google.registry.model.server.KmsSecretRevision</class>
     <class>google.registry.model.server.Lock</class>
     <class>google.registry.model.server.ServerSecret</class>
     <class>google.registry.model.smd.SignedMarkRevocationList</class>

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

@@ -18,6 +18,7 @@ import static com.google.common.collect.Iterables.transform;
 import static com.google.common.truth.Truth.assertThat;
 import static google.registry.backup.BackupUtils.GcsMetadataKeys.LOWER_BOUND_CHECKPOINT;
 import static google.registry.backup.ExportCommitLogDiffAction.DIFF_FILE_PREFIX;
+import static google.registry.backup.GcsDiffFileLister.getCommitLogDiffPrefix;
 import static org.joda.time.DateTimeZone.UTC;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.fail;
@@ -61,7 +62,7 @@ public class GcsDiffFileListerTest {
   @BeforeEach
   void beforeEach() throws Exception {
     diffLister.gcsUtils = gcsUtils;
-    diffLister.lazyExecutor = MoreExecutors::newDirectExecutorService;
+    diffLister.executorProvider = MoreExecutors::newDirectExecutorService;
     diffLister.scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
     for (int i = 0; i < 5; i++) {
       addGcsFile(i, i + 1);
@@ -189,12 +190,23 @@ public class GcsDiffFileListerTest {
 
   @Test
   void testList_toTimeSpecified() {
-    assertThat(listDiffFiles(
-            now.minusMinutes(4).minusSeconds(1), now.minusMinutes(2).plusSeconds(1)))
-        .containsExactly(
-            now.minusMinutes(4),
-            now.minusMinutes(3),
-            now.minusMinutes(2))
+    assertThat(
+            listDiffFiles(now.minusMinutes(4).minusSeconds(1), now.minusMinutes(2).plusSeconds(1)))
+        .containsExactly(now.minusMinutes(4), now.minusMinutes(3), now.minusMinutes(2))
         .inOrder();
   }
+
+  @Test
+  void testPrefix_lengthened() {
+    DateTime from = DateTime.parse("2021-05-11T06:48:00.070Z");
+    assertThat(getCommitLogDiffPrefix(from, null)).isEqualTo("commit_diff_until_");
+    assertThat(getCommitLogDiffPrefix(from, DateTime.parse("2021-07-01")))
+        .isEqualTo("commit_diff_until_2021-");
+    assertThat(getCommitLogDiffPrefix(from, DateTime.parse("2021-05-21")))
+        .isEqualTo("commit_diff_until_2021-05-");
+    assertThat(getCommitLogDiffPrefix(from, DateTime.parse("2021-05-11T09:48:00.070Z")))
+        .isEqualTo("commit_diff_until_2021-05-11T");
+    assertThat(getCommitLogDiffPrefix(from, DateTime.parse("2021-05-11T06:59:00.070Z")))
+        .isEqualTo("commit_diff_until_2021-05-11T06:");
+  }
 }

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

@@ -67,6 +67,7 @@ import google.registry.persistence.transaction.TransactionManagerFactory;
 import google.registry.schema.replay.SqlReplayCheckpoint;
 import google.registry.schema.tld.PremiumEntry;
 import google.registry.testing.AppEngineExtension;
+import google.registry.testing.DatabaseHelper;
 import google.registry.testing.FakeClock;
 import google.registry.testing.FakeResponse;
 import google.registry.testing.TestObject;
@@ -132,7 +133,7 @@ public class ReplayCommitLogsToSqlActionTest {
     action.gcsBucket = "gcs bucket";
     action.diffLister = new GcsDiffFileLister();
     action.diffLister.gcsUtils = gcsUtils;
-    action.diffLister.lazyExecutor = MoreExecutors::newDirectExecutorService;
+    action.diffLister.executorProvider = MoreExecutors::newDirectExecutorService;
     action.diffLister.scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
     ofyTm()
         .transact(
@@ -149,7 +150,7 @@ public class ReplayCommitLogsToSqlActionTest {
 
   @AfterEach
   void afterEach() {
-    ofyTm().transact(() -> DatabaseMigrationStateSchedule.set(DEFAULT_TRANSITION_MAP.toValueMap()));
+    DatabaseHelper.removeDatabaseMigrationSchedule();
   }
 
   @Test
@@ -200,7 +201,7 @@ public class ReplayCommitLogsToSqlActionTest {
         CommitLogMutation.create(manifest2Key, TestObject.create("f")));
     jpaTm().transact(() -> SqlReplayCheckpoint.set(now.minusMinutes(1).minusMillis(1)));
     fakeClock.advanceOneMilli();
-    runAndAssertSuccess(now);
+    runAndAssertSuccess(now, 2);
     assertExpectedIds("previous to keep", "b", "d", "e", "f");
   }
 
@@ -211,7 +212,7 @@ public class ReplayCommitLogsToSqlActionTest {
     saveDiffFileNotToRestore(gcsUtils, now.minusMinutes(1));
     saveDiffFile(gcsUtils, createCheckpoint(now.minusMillis(2)));
     jpaTm().transact(() -> SqlReplayCheckpoint.set(now.minusMillis(1)));
-    runAndAssertSuccess(now.minusMillis(1));
+    runAndAssertSuccess(now.minusMillis(1), 0);
     assertExpectedIds("previous to keep");
   }
 
@@ -235,8 +236,10 @@ public class ReplayCommitLogsToSqlActionTest {
     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");
+            "Running in dry-run mode, the first set of commit log files processed would be from "
+                + "searching from 1999-12-31T23:59:00.000Z to 1999-12-31T23:59:59.999Z and would "
+                + "contain 1 file(s). They are (limit 10): \n"
+                + "[commit_diff_until_1999-12-31T23:59:00.000Z]");
   }
 
   @Test
@@ -253,7 +256,7 @@ public class ReplayCommitLogsToSqlActionTest {
         CommitLogManifest.create(bucketKey, now, null),
         CommitLogMutation.create(manifestKey, TestObject.create("a")),
         CommitLogMutation.create(manifestKey, TestObject.create("b")));
-    runAndAssertSuccess(now.minusMinutes(1));
+    runAndAssertSuccess(now.minusMinutes(1), 1);
     assertExpectedIds("previous to keep", "a", "b");
   }
 
@@ -275,7 +278,7 @@ public class ReplayCommitLogsToSqlActionTest {
             getBucketKey(1),
             now,
             ImmutableSet.of(Key.create(TestObject.create("previous to delete")))));
-    runAndAssertSuccess(now.minusMinutes(1));
+    runAndAssertSuccess(now.minusMinutes(1), 1);
     assertExpectedIds("previous to keep");
   }
 
@@ -347,7 +350,7 @@ public class ReplayCommitLogsToSqlActionTest {
         domainMutation,
         contactMutation);
 
-    runAndAssertSuccess(now.minusMinutes(1));
+    runAndAssertSuccess(now.minusMinutes(1), 1);
     // Verify two things:
     // 1. that the contact insert occurred before the domain insert (necessary for FK ordering)
     //    even though the domain came first in the file
@@ -390,7 +393,7 @@ public class ReplayCommitLogsToSqlActionTest {
         CommitLogManifest.create(
             getBucketKey(1), now.minusMinutes(1).plusMillis(1), ImmutableSet.of()),
         contactMutation);
-    runAndAssertSuccess(now.minusMinutes(1).plusMillis(1));
+    runAndAssertSuccess(now.minusMinutes(1).plusMillis(1), 1);
     // Verify that the delete occurred first (because it was in the first transaction) even though
     // deletes have higher weight
     ArgumentCaptor<Object> putCaptor = ArgumentCaptor.forClass(Object.class);
@@ -435,7 +438,7 @@ public class ReplayCommitLogsToSqlActionTest {
                 throw new RuntimeException(e);
               }
             });
-    runAndAssertSuccess(now.minusMinutes(1));
+    runAndAssertSuccess(now.minusMinutes(1), 1);
     // jpaTm()::put should only have been called with the checkpoint
     verify(spy, times(2)).put(any(SqlReplayCheckpoint.class));
     verify(spy, times(2)).put(any());
@@ -460,7 +463,7 @@ public class ReplayCommitLogsToSqlActionTest {
             // one object only exists in Datastore, one is dually-written (so isn't replicated)
             ImmutableSet.of(getCrossTldKey(), claimsListKey)));
 
-    runAndAssertSuccess(now.minusMinutes(1));
+    runAndAssertSuccess(now.minusMinutes(1), 1);
     verify(spy, times(0)).delete(any(VKey.class));
   }
 
@@ -503,7 +506,7 @@ public class ReplayCommitLogsToSqlActionTest {
         createCheckpoint(now.minusMinutes(1)),
         CommitLogManifest.create(bucketKey, now, null),
         CommitLogMutation.create(manifestKey, TestObject.create("a")));
-    runAndAssertSuccess(now.minusMinutes(1));
+    runAndAssertSuccess(now.minusMinutes(1), 1);
     assertThat(TestObject.beforeSqlSaveCallCount).isEqualTo(1);
   }
 
@@ -522,11 +525,12 @@ public class ReplayCommitLogsToSqlActionTest {
     assertThat(TestObject.beforeSqlDeleteCallCount).isEqualTo(1);
   }
 
-  private void runAndAssertSuccess(DateTime expectedCheckpointTime) {
+  private void runAndAssertSuccess(DateTime expectedCheckpointTime, int numFiles) {
     action.run();
     assertThat(response.getStatus()).isEqualTo(SC_OK);
     assertThat(response.getPayload())
-        .isEqualTo("ReplayCommitLogsToSqlAction completed successfully.");
+        .isEqualTo(
+            String.format("Caught up to current time after replaying %d file(s).", numFiles));
     assertThat(jpaTm().transact(SqlReplayCheckpoint::get)).isEqualTo(expectedCheckpointTime);
   }
 

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

@@ -88,7 +88,7 @@ public class RestoreCommitLogsActionTest {
     action.gcsBucketOverride = Optional.empty();
     action.diffLister = new GcsDiffFileLister();
     action.diffLister.gcsUtils = gcsUtils;
-    action.diffLister.lazyExecutor = MoreExecutors::newDirectExecutorService;
+    action.diffLister.executorProvider = MoreExecutors::newDirectExecutorService;
     action.diffLister.scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
   }
 

core/src/test/java/google/registry/beam/rde/RdePipelineTest.java

@@ -23,7 +23,6 @@ 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.setTmForTest;
 import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
 import static google.registry.rde.RdeResourceType.CONTACT;
 import static google.registry.rde.RdeResourceType.DOMAIN;
@@ -50,7 +49,7 @@ 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.persistence.transaction.TransactionManagerFactory;
 import google.registry.rde.DepositFragment;
 import google.registry.rde.PendingDeposit;
 import google.registry.rde.RdeResourceType;
@@ -117,12 +116,9 @@ public class RdePipelineTest {
 
   private RdePipeline rdePipeline;
 
-  private TransactionManager originalTm;
-
   @BeforeEach
   void beforeEach() throws Exception {
-    originalTm = tm();
-    setTmForTest(jpaTm());
+    TransactionManagerFactory.setTmForTest(jpaTm());
     loadInitialData();
 
     // Two real registrars have been created by loadInitialData(), named "New Registrar" and "The
@@ -169,7 +165,7 @@ public class RdePipelineTest {
 
   @AfterEach
   void afterEach() {
-    setTmForTest(originalTm);
+    TransactionManagerFactory.removeTmOverrideForTest();
   }
 
   @Test

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

@@ -18,8 +18,8 @@ import static com.google.common.collect.ImmutableList.toImmutableList;
 import static com.google.common.truth.Truth.assertThat;
 import static google.registry.model.ImmutableObjectSubject.immutableObjectCorrespondence;
 import static google.registry.persistence.transaction.TransactionManagerFactory.jpaTm;
+import static google.registry.persistence.transaction.TransactionManagerFactory.removeTmOverrideForTest;
 import static google.registry.persistence.transaction.TransactionManagerFactory.setTmForTest;
-import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
 import static google.registry.testing.AppEngineExtension.makeRegistrar1;
 import static google.registry.testing.DatabaseHelper.createTld;
 import static google.registry.testing.DatabaseHelper.persistActiveContact;
@@ -49,7 +49,6 @@ import google.registry.model.reporting.Spec11ThreatMatch.ThreatType;
 import google.registry.model.reporting.Spec11ThreatMatchDao;
 import google.registry.persistence.transaction.JpaTestRules;
 import google.registry.persistence.transaction.JpaTestRules.JpaIntegrationTestExtension;
-import google.registry.persistence.transaction.TransactionManager;
 import google.registry.testing.DatastoreEntityExtension;
 import google.registry.testing.FakeClock;
 import google.registry.testing.FakeSleeper;
@@ -230,7 +229,6 @@ class Spec11PipelineTest {
   }
 
   private void setupCloudSql() {
-    TransactionManager originalTm = tm();
     setTmForTest(jpaTm());
     persistNewRegistrar("TheRegistrar");
     persistNewRegistrar("NewRegistrar");
@@ -271,7 +269,7 @@ class Spec11PipelineTest {
     persistResource(createDomain("no-email.com", "2A4BA9BBC-COM", registrar2, contact2));
     persistResource(
         createDomain("anti-anti-anti-virus.dev", "555666888-DEV", registrar3, contact3));
-    setTmForTest(originalTm);
+    removeTmOverrideForTest();
   }
 
   private void verifySaveToGcs() throws Exception {

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

@@ -55,6 +55,7 @@ class EppLoginTlsTest extends EppTestCase {
       new CertificateChecker(
           ImmutableSortedMap.of(START_OF_TIME, 825, DateTime.parse("2020-09-01T00:00:00Z"), 398),
           30,
+          15,
           2048,
           ImmutableSet.of("secp256r1", "secp384r1"),
           clock);

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

@@ -64,6 +64,7 @@ class FlowRunnerTest {
       new CertificateChecker(
           ImmutableSortedMap.of(START_OF_TIME, 825, DateTime.parse("2020-09-01T00:00:00Z"), 398),
           30,
+          15,
           2048,
           ImmutableSet.of("secp256r1", "secp384r1"),
           clock);

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

@@ -53,6 +53,7 @@ final class TlsCredentialsTest {
       new CertificateChecker(
           ImmutableSortedMap.of(START_OF_TIME, 825, DateTime.parse("2020-09-01T00:00:00Z"), 398),
           30,
+          15,
           2048,
           ImmutableSet.of("secp256r1", "secp384r1"),
           clock);

core/src/test/java/google/registry/flows/certs/CertificateCheckerTest.java

@@ -50,6 +50,7 @@ class CertificateCheckerTest {
       new CertificateChecker(
           ImmutableSortedMap.of(START_OF_TIME, 825, DateTime.parse("2020-09-01T00:00:00Z"), 398),
           30,
+          15,
           2048,
           ImmutableSet.of("secp256r1", "secp384r1"),
           fakeClock);
@@ -214,23 +215,45 @@ class CertificateCheckerTest {
         assertThrows(
             IllegalArgumentException.class,
             () -> certificateChecker.checkCertificate("bad certificate string"));
-    assertThat(thrown).hasMessageThat().isEqualTo("Unable to read given certificate.");
+    assertThat(thrown).hasMessageThat().contains("Unable to read given certificate");
   }
 
   @Test
-  void test_isNearingExpiration_yesItIs() throws Exception {
-    fakeClock.setTo(DateTime.parse("2021-09-20T00:00:00Z"));
+  void test_getCertificate_throwsException_invalidCertificateString() {
+    IllegalArgumentException thrown =
+        assertThrows(
+            IllegalArgumentException.class,
+            () -> certificateChecker.getCertificate("invalidCertificateString"));
+    assertThat(thrown).hasMessageThat().contains("Unable to read given certificate");
+  }
+
+  @Test
+  void test_getCertificate_returnsCertificateObject() throws Exception {
     X509Certificate certificate =
         SelfSignedCaCertificate.create(
                 SSL_HOST,
                 DateTime.parse("2020-09-02T00:00:00Z"),
                 DateTime.parse("2021-10-01T00:00:00Z"))
             .cert();
-    assertThat(certificateChecker.isNearingExpiration(certificate)).isTrue();
+    String certificateStr = certificateChecker.serializeCertificate(certificate);
+    assertThat(certificateChecker.getCertificate(certificateStr).equals(certificate)).isTrue();
   }
 
   @Test
-  void test_isNearingExpiration_noItsNot() throws Exception {
+  void test_serializeCertificate_returnsCertificateString() throws Exception {
+    X509Certificate certificate =
+        SelfSignedCaCertificate.create(
+                SSL_HOST,
+                DateTime.parse("2020-09-02T00:00:00Z"),
+                DateTime.parse("2021-10-01T00:00:00Z"))
+            .cert();
+    assertThat(
+            certificateChecker.getCertificate(certificateChecker.serializeCertificate(certificate)))
+        .isEqualTo(certificate);
+  }
+
+  @Test
+  void test_shouldReceiveExpiringNotification_returnsFalse_greaterThan30() throws Exception {
     fakeClock.setTo(DateTime.parse("2021-07-20T00:00:00Z"));
     X509Certificate certificate =
         SelfSignedCaCertificate.create(
@@ -238,7 +261,124 @@ class CertificateCheckerTest {
                 DateTime.parse("2020-09-02T00:00:00Z"),
                 DateTime.parse("2021-10-01T00:00:00Z"))
             .cert();
-    assertThat(certificateChecker.isNearingExpiration(certificate)).isFalse();
+    String certificateStr = certificateChecker.serializeCertificate(certificate);
+    assertThat(certificateChecker.shouldReceiveExpiringNotification(null, certificateStr))
+        .isFalse();
+  }
+
+  @Test
+  void test_shouldReceiveExpiringNotification_returnsFalse_hasExpired() throws Exception {
+    fakeClock.setTo(DateTime.parse("2021-10-20T00:00:00Z"));
+    X509Certificate certificate =
+        SelfSignedCaCertificate.create(
+                SSL_HOST,
+                DateTime.parse("2020-09-02T00:00:00Z"),
+                DateTime.parse("2021-10-01T00:00:00Z"))
+            .cert();
+    String certificateStr = certificateChecker.serializeCertificate(certificate);
+    assertThat(certificateChecker.shouldReceiveExpiringNotification(null, certificateStr))
+        .isFalse();
+  }
+
+  @Test
+  void test_shouldReceiveExpiringNotification_returnsTrue_on15days() throws Exception {
+    fakeClock.setTo(DateTime.parse("2021-08-16T00:00:00Z"));
+    X509Certificate certificate =
+        SelfSignedCaCertificate.create(
+                SSL_HOST,
+                DateTime.parse("2020-09-02T00:00:00Z"),
+                DateTime.parse("2021-08-30T00:00:00Z"))
+            .cert();
+    String certificateStr = certificateChecker.serializeCertificate(certificate);
+    DateTime lastExpiringNotificationSentDate = DateTime.parse("2021-08-01T00:00:00Z");
+    assertThat(
+            certificateChecker.shouldReceiveExpiringNotification(
+                lastExpiringNotificationSentDate, certificateStr))
+        .isTrue();
+  }
+
+  @Test
+  void test_shouldReceiveExpiringNotification_returnsTrue_on30days() throws Exception {
+    fakeClock.setTo(DateTime.parse("2021-01-01T00:00:00Z"));
+    X509Certificate certificate =
+        SelfSignedCaCertificate.create(
+                SSL_HOST,
+                DateTime.parse("2020-01-02T00:00:00Z"),
+                DateTime.parse("2021-01-30T00:00:00Z"))
+            .cert();
+    String certificateStr = certificateChecker.serializeCertificate(certificate);
+    assertThat(certificateChecker.shouldReceiveExpiringNotification(null, certificateStr)).isTrue();
+  }
+
+  @Test
+  void test_shouldReceiveExpiringNotification_returnsFalse_between30and15_lastSentDateIsNotNull()
+      throws Exception {
+    fakeClock.setTo(DateTime.parse("2021-07-05T00:00:00Z"));
+    X509Certificate certificate =
+        SelfSignedCaCertificate.create(
+                SSL_HOST,
+                DateTime.parse("2020-06-02T00:00:00Z"),
+                DateTime.parse("2021-07-25T00:00:00Z"))
+            .cert();
+    String certificateStr = certificateChecker.serializeCertificate(certificate);
+    DateTime lastExpiringNotificationSentDate = DateTime.parse("2021-07-04T00:00:00Z");
+    assertThat(
+            certificateChecker.shouldReceiveExpiringNotification(
+                lastExpiringNotificationSentDate, certificateStr))
+        .isFalse();
+  }
+
+  @Test
+  void test_shouldReceiveExpiringNotification_returnsTrue_between30and15_lastSentDateIsNull()
+      throws Exception {
+    fakeClock.setTo(DateTime.parse("2021-07-05T00:00:00Z"));
+    X509Certificate certificate =
+        SelfSignedCaCertificate.create(
+                SSL_HOST,
+                DateTime.parse("2020-06-02T00:00:00Z"),
+                DateTime.parse("2021-07-25T00:00:00Z"))
+            .cert();
+    String certificateStr = certificateChecker.serializeCertificate(certificate);
+    assertThat(certificateChecker.shouldReceiveExpiringNotification(null, certificateStr)).isTrue();
+  }
+
+  @Test
+  void
+      test_shouldReceiveExpiringNotification_returnsFalse_between15and0_lastSentDateBetween30and15_butLate()
+          throws Exception {
+    fakeClock.setTo(DateTime.parse("2021-07-05T00:00:00Z"));
+    X509Certificate certificate =
+        SelfSignedCaCertificate.create(
+                SSL_HOST,
+                DateTime.parse("2020-06-02T00:00:00Z"),
+                DateTime.parse("2021-07-18T00:00:00Z"))
+            .cert();
+    String certificateStr = certificateChecker.serializeCertificate(certificate);
+    DateTime lastExpiringNotificationSentDate = DateTime.parse("2021-07-01T00:00:00Z");
+    // The first notification date is late and difference between now and
+    // lastExpiringNotificationSentDate is within expirationWarningIntervalDays.
+    assertThat(
+            certificateChecker.shouldReceiveExpiringNotification(
+                lastExpiringNotificationSentDate, certificateStr))
+        .isFalse();
+  }
+
+  @Test
+  void test_shouldReceiveExpiringNotification_returnsTrue_between15and0_lastSentDateBetween30and15()
+      throws Exception {
+    fakeClock.setTo(DateTime.parse("2021-07-05T00:00:00Z"));
+    X509Certificate certificate =
+        SelfSignedCaCertificate.create(
+                SSL_HOST,
+                DateTime.parse("2020-06-02T00:00:00Z"),
+                DateTime.parse("2021-07-18T00:00:00Z"))
+            .cert();
+    String certificateStr = certificateChecker.serializeCertificate(certificate);
+    DateTime lastExpiringNotificationSentDate = DateTime.parse("2021-06-20T00:00:00Z");
+    assertThat(
+            certificateChecker.shouldReceiveExpiringNotification(
+                lastExpiringNotificationSentDate, certificateStr))
+        .isTrue();
   }
 
   @Test

core/src/test/java/google/registry/flows/session/LoginFlowViaTlsTest.java

@@ -50,6 +50,7 @@ public class LoginFlowViaTlsTest extends LoginFlowTestCase {
       new CertificateChecker(
           ImmutableSortedMap.of(START_OF_TIME, 825, DateTime.parse("2020-09-01T00:00:00Z"), 398),
           30,
+          15,
           2048,
           ImmutableSet.of("secp256r1", "secp384r1"),
           clock);

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

@@ -15,7 +15,6 @@
 package google.registry.model.common;
 
 import static com.google.common.truth.Truth.assertThat;
-import static google.registry.model.common.DatabaseMigrationStateSchedule.DEFAULT_TRANSITION_MAP;
 import static google.registry.model.common.DatabaseMigrationStateSchedule.MigrationState.DATASTORE_ONLY;
 import static google.registry.model.common.DatabaseMigrationStateSchedule.MigrationState.DATASTORE_PRIMARY;
 import static google.registry.model.common.DatabaseMigrationStateSchedule.MigrationState.DATASTORE_PRIMARY_READ_ONLY;
@@ -30,6 +29,7 @@ import static org.junit.Assert.assertThrows;
 import com.google.common.collect.ImmutableSortedMap;
 import google.registry.model.EntityTestCase;
 import google.registry.model.common.DatabaseMigrationStateSchedule.MigrationState;
+import google.registry.testing.DatabaseHelper;
 import org.joda.time.DateTime;
 import org.joda.time.Duration;
 import org.junit.jupiter.api.AfterEach;
@@ -45,7 +45,7 @@ public class DatabaseMigrationStateScheduleTest extends EntityTestCase {
 
   @AfterEach
   void afterEach() {
-    ofyTm().transact(() -> DatabaseMigrationStateSchedule.set(DEFAULT_TRANSITION_MAP.toValueMap()));
+    DatabaseHelper.removeDatabaseMigrationSchedule();
   }
 
   @Test

core/src/test/java/google/registry/model/history/ContactHistoryTest.java

@@ -18,6 +18,7 @@ import static com.google.common.truth.Truth.assertThat;
 import static google.registry.model.ImmutableObjectSubject.assertAboutImmutableObjects;
 import static google.registry.persistence.transaction.TransactionManagerFactory.jpaTm;
 import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
+import static google.registry.testing.DatabaseHelper.newContactResource;
 import static google.registry.testing.DatabaseHelper.newContactResourceWithRoid;
 import static google.registry.testing.SqlHelper.saveRegistrar;
 import static java.nio.charset.StandardCharsets.UTF_8;
@@ -47,7 +48,7 @@ public class ContactHistoryTest extends EntityTestCase {
     jpaTm().transact(() -> jpaTm().insert(contact));
     VKey<ContactResource> contactVKey = contact.createVKey();
     ContactResource contactFromDb = jpaTm().transact(() -> jpaTm().loadByKey(contactVKey));
-    ContactHistory contactHistory = createContactHistory(contactFromDb, contact.getRepoId());
+    ContactHistory contactHistory = createContactHistory(contactFromDb);
     jpaTm().transact(() -> jpaTm().insert(contactHistory));
     jpaTm()
         .transact(
@@ -67,10 +68,7 @@ public class ContactHistoryTest extends EntityTestCase {
     VKey<ContactResource> contactVKey = contact.createVKey();
     ContactResource contactFromDb = jpaTm().transact(() -> jpaTm().loadByKey(contactVKey));
     ContactHistory contactHistory =
-        createContactHistory(contactFromDb, contact.getRepoId())
-            .asBuilder()
-            .setContact(null)
-            .build();
+        createContactHistory(contactFromDb).asBuilder().setContact(null).build();
     jpaTm().transact(() -> jpaTm().insert(contactHistory));
 
     jpaTm()
@@ -91,7 +89,7 @@ public class ContactHistoryTest extends EntityTestCase {
     VKey<ContactResource> contactVKey = contact.createVKey();
     ContactResource contactFromDb = tm().transact(() -> tm().loadByKey(contactVKey));
     fakeClock.advanceOneMilli();
-    ContactHistory contactHistory = createContactHistory(contactFromDb, contact.getRepoId());
+    ContactHistory contactHistory = createContactHistory(contactFromDb);
     tm().transact(() -> tm().insert(contactHistory));
 
     // retrieving a HistoryEntry or a ContactHistory with the same key should return the same object
@@ -106,7 +104,38 @@ public class ContactHistoryTest extends EntityTestCase {
     assertThat(hostHistoryFromDb).isEqualTo(historyEntryFromDb);
   }
 
-  private ContactHistory createContactHistory(ContactBase contact, String contactRepoId) {
+  @Test
+  void testBeforeSqlSave_afterContactPersisted() {
+    saveRegistrar("TheRegistrar");
+    ContactResource contactResource = newContactResource("contactId");
+    ContactHistory contactHistory =
+        new ContactHistory.Builder()
+            .setType(HistoryEntry.Type.HOST_CREATE)
+            .setXmlBytes("<xml></xml>".getBytes(UTF_8))
+            .setModificationTime(fakeClock.nowUtc())
+            .setClientId("TheRegistrar")
+            .setTrid(Trid.create("ABC-123", "server-trid"))
+            .setBySuperuser(false)
+            .setReason("reason")
+            .setRequestedByRegistrar(true)
+            .setContactRepoId(contactResource.getRepoId())
+            .build();
+    jpaTm()
+        .transact(
+            () -> {
+              jpaTm().put(contactResource);
+              contactHistory.beforeSqlSaveOnReplay();
+              jpaTm().put(contactHistory);
+            });
+    jpaTm()
+        .transact(
+            () ->
+                assertAboutImmutableObjects()
+                    .that(jpaTm().loadByEntity(contactResource))
+                    .hasFieldsEqualTo(jpaTm().loadByEntity(contactHistory).getContactBase().get()));
+  }
+
+  private ContactHistory createContactHistory(ContactBase contact) {
     return new ContactHistory.Builder()
         .setType(HistoryEntry.Type.HOST_CREATE)
         .setXmlBytes("<xml></xml>".getBytes(UTF_8))
@@ -117,7 +146,7 @@ public class ContactHistoryTest extends EntityTestCase {
         .setReason("reason")
         .setRequestedByRegistrar(true)
         .setContact(contact)
-        .setContactRepoId(contactRepoId)
+        .setContactRepoId(contact.getRepoId())
         .build();
   }
 

core/src/test/java/google/registry/model/history/DomainHistoryTest.java

@@ -190,6 +190,44 @@ public class DomainHistoryTest extends EntityTestCase {
     jpaTm().transact(() -> jpaTm().put(domainHistoryFromDb2));
   }
 
+  @TestSqlOnly
+  void testBeforeSqlSave_afterDomainPersisted() {
+    DomainBase domain = createDomainWithContactsAndHosts();
+    DomainHistory historyWithoutResource =
+        new DomainHistory.Builder()
+            .setType(HistoryEntry.Type.DOMAIN_CREATE)
+            .setXmlBytes("<xml></xml>".getBytes(UTF_8))
+            .setModificationTime(fakeClock.nowUtc())
+            .setClientId("TheRegistrar")
+            .setTrid(Trid.create("ABC-123", "server-trid"))
+            .setBySuperuser(false)
+            .setReason("reason")
+            .setRequestedByRegistrar(true)
+            .setDomainRepoId(domain.getRepoId())
+            .setOtherClientId("otherClient")
+            .setPeriod(Period.create(1, Period.Unit.YEARS))
+            .build();
+    jpaTm()
+        .transact(
+            () -> {
+              jpaTm()
+                  .put(
+                      domain
+                          .asBuilder()
+                          .setPersistedCurrentSponsorClientId("NewRegistrar")
+                          .build());
+              historyWithoutResource.beforeSqlSaveOnReplay();
+              jpaTm().put(historyWithoutResource);
+            });
+    jpaTm()
+        .transact(
+            () ->
+                assertAboutImmutableObjects()
+                    .that(jpaTm().loadByEntity(domain))
+                    .hasFieldsEqualTo(
+                        jpaTm().loadByEntity(historyWithoutResource).getDomainContent().get()));
+  }
+
   static DomainBase createDomainWithContactsAndHosts() {
     createTld("tld");
     HostResource host = newHostResourceWithRoid("ns1.example.com", "host1");

core/src/test/java/google/registry/model/history/HostHistoryTest.java

@@ -18,6 +18,7 @@ import static com.google.common.truth.Truth.assertThat;
 import static google.registry.model.ImmutableObjectSubject.assertAboutImmutableObjects;
 import static google.registry.persistence.transaction.TransactionManagerFactory.jpaTm;
 import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
+import static google.registry.testing.DatabaseHelper.newHostResource;
 import static google.registry.testing.DatabaseHelper.newHostResourceWithRoid;
 import static google.registry.testing.SqlHelper.saveRegistrar;
 import static java.nio.charset.StandardCharsets.UTF_8;
@@ -48,7 +49,7 @@ public class HostHistoryTest extends EntityTestCase {
     VKey<HostResource> hostVKey =
         VKey.create(HostResource.class, "host1", Key.create(HostResource.class, "host1"));
     HostResource hostFromDb = jpaTm().transact(() -> jpaTm().loadByKey(hostVKey));
-    HostHistory hostHistory = createHostHistory(hostFromDb, host.getRepoId());
+    HostHistory hostHistory = createHostHistory(hostFromDb);
     jpaTm().transact(() -> jpaTm().insert(hostHistory));
     jpaTm()
         .transact(
@@ -66,8 +67,7 @@ public class HostHistoryTest extends EntityTestCase {
     jpaTm().transact(() -> jpaTm().insert(host));
 
     HostResource hostFromDb = jpaTm().transact(() -> jpaTm().loadByKey(host.createVKey()));
-    HostHistory hostHistory =
-        createHostHistory(hostFromDb, host.getRepoId()).asBuilder().setHost(null).build();
+    HostHistory hostHistory = createHostHistory(hostFromDb).asBuilder().setHost(null).build();
     jpaTm().transact(() -> jpaTm().insert(hostHistory));
 
     jpaTm()
@@ -88,7 +88,7 @@ public class HostHistoryTest extends EntityTestCase {
     VKey<HostResource> hostVKey =
         VKey.create(HostResource.class, "host1", Key.create(HostResource.class, "host1"));
     HostResource hostFromDb = tm().transact(() -> tm().loadByKey(hostVKey));
-    HostHistory hostHistory = createHostHistory(hostFromDb, host.getRepoId());
+    HostHistory hostHistory = createHostHistory(hostFromDb);
     fakeClock.advanceOneMilli();
     tm().transact(() -> tm().insert(hostHistory));
 
@@ -104,6 +104,37 @@ public class HostHistoryTest extends EntityTestCase {
     assertThat(hostHistoryFromDb).isEqualTo(historyEntryFromDb);
   }
 
+  @Test
+  void testBeforeSqlSave_afterHostPersisted() {
+    saveRegistrar("TheRegistrar");
+    HostResource hostResource = newHostResource("ns1.example.tld");
+    HostHistory hostHistory =
+        new HostHistory.Builder()
+            .setType(HistoryEntry.Type.HOST_CREATE)
+            .setXmlBytes("<xml></xml>".getBytes(UTF_8))
+            .setModificationTime(fakeClock.nowUtc())
+            .setClientId("TheRegistrar")
+            .setTrid(Trid.create("ABC-123", "server-trid"))
+            .setBySuperuser(false)
+            .setReason("reason")
+            .setRequestedByRegistrar(true)
+            .setHostRepoId(hostResource.getRepoId())
+            .build();
+    jpaTm()
+        .transact(
+            () -> {
+              jpaTm().put(hostResource);
+              hostHistory.beforeSqlSaveOnReplay();
+              jpaTm().put(hostHistory);
+            });
+    jpaTm()
+        .transact(
+            () ->
+                assertAboutImmutableObjects()
+                    .that(jpaTm().loadByEntity(hostResource))
+                    .hasFieldsEqualTo(jpaTm().loadByEntity(hostHistory).getHostBase().get()));
+  }
+
   private void assertHostHistoriesEqual(HostHistory one, HostHistory two) {
     assertAboutImmutableObjects().that(one).isEqualExceptFields(two, "hostBase");
     assertAboutImmutableObjects()
@@ -111,7 +142,7 @@ public class HostHistoryTest extends EntityTestCase {
         .isEqualExceptFields(two.getHostBase().orElse(null), "repoId");
   }
 
-  private HostHistory createHostHistory(HostBase hostBase, String hostRepoId) {
+  private HostHistory createHostHistory(HostBase hostBase) {
     return new HostHistory.Builder()
         .setType(HistoryEntry.Type.HOST_CREATE)
         .setXmlBytes("<xml></xml>".getBytes(UTF_8))
@@ -122,7 +153,7 @@ public class HostHistoryTest extends EntityTestCase {
         .setReason("reason")
         .setRequestedByRegistrar(true)
         .setHost(hostBase)
-        .setHostRepoId(hostRepoId)
+        .setHostRepoId(hostBase.getRepoId())
         .build();
   }
 }

core/src/test/java/google/registry/model/history/LegacyHistoryObjectTest.java

@@ -74,7 +74,7 @@ public class LegacyHistoryObjectTest extends EntityTestCase {
     assertAboutImmutableObjects()
         .that(legacyHistoryEntry)
         .isEqualExceptFields(fromObjectify, "contactBase", "contactRepoId");
-    assertThat(fromObjectify instanceof ContactHistory).isTrue();
+    assertThat(fromObjectify).isInstanceOf(ContactHistory.class);
     ContactHistory legacyContactHistory = (ContactHistory) fromObjectify;
 
     // Next, save that from-Datastore object in SQL and verify we can load it back in
@@ -124,7 +124,7 @@ public class LegacyHistoryObjectTest extends EntityTestCase {
             "nsHosts",
             "dsDataHistories",
             "gracePeriodHistories");
-    assertThat(fromObjectify instanceof DomainHistory).isTrue();
+    assertThat(fromObjectify).isInstanceOf(DomainHistory.class);
     DomainHistory legacyDomainHistory = (DomainHistory) fromObjectify;
 
     // Next, save that from-Datastore object in SQL and verify we can load it back in
@@ -137,15 +137,7 @@ public class LegacyHistoryObjectTest extends EntityTestCase {
               // Don't compare nsHosts directly because one is null and the other is empty
               assertAboutImmutableObjects()
                   .that(legacyDomainHistory)
-                  .isEqualExceptFields(
-                      // NB: period, transaction records, and other client ID are added in #794
-                      legacyHistoryFromSql,
-                      "period",
-                      "domainTransactionRecords",
-                      "otherClientId",
-                      "nsHosts",
-                      "dsDataHistories",
-                      "gracePeriodHistories");
+                  .isEqualExceptFields(legacyHistoryFromSql, "nsHosts");
               assertThat(nullToEmpty(legacyDomainHistory.getNsHosts()))
                   .isEqualTo(nullToEmpty(legacyHistoryFromSql.getNsHosts()));
             });
@@ -174,7 +166,7 @@ public class LegacyHistoryObjectTest extends EntityTestCase {
     assertAboutImmutableObjects()
         .that(legacyHistoryEntry)
         .isEqualExceptFields(fromObjectify, "hostBase", "hostRepoId");
-    assertThat(fromObjectify instanceof HostHistory).isTrue();
+    assertThat(fromObjectify).isInstanceOf(HostHistory.class);
     HostHistory legacyHostHistory = (HostHistory) fromObjectify;
 
     // Next, save that from-Datastore object in SQL and verify we can load it back in

core/src/test/java/google/registry/model/registrar/RegistrarTest.java

@@ -29,6 +29,7 @@ import static google.registry.testing.DatabaseHelper.newRegistry;
 import static google.registry.testing.DatabaseHelper.persistResource;
 import static google.registry.testing.DatabaseHelper.persistSimpleResource;
 import static google.registry.testing.DatabaseHelper.persistSimpleResources;
+import static google.registry.util.DateTimeUtils.START_OF_TIME;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 
 import com.google.common.collect.ImmutableList;
@@ -414,6 +415,64 @@ class RegistrarTest extends EntityTestCase {
         IllegalArgumentException.class, () -> new Registrar.Builder().setPhonePasscode("code1"));
   }
 
+  @TestOfyAndSql
+  void testSuccess_getLastExpiringCertNotificationSentDate_returnsInitialValue() {
+    assertThat(registrar.getLastExpiringCertNotificationSentDate()).isEqualTo(START_OF_TIME);
+  }
+
+  @TestOfyAndSql
+  void testSuccess_getLastExpiringFailoverCertNotificationSentDate_returnsInitialValue() {
+    assertThat(registrar.getLastExpiringFailoverCertNotificationSentDate())
+        .isEqualTo(START_OF_TIME);
+  }
+
+  @TestOfyAndSql
+  void testSuccess_setLastExpiringCertNotificationSentDate() {
+    assertThat(
+            registrar
+                .asBuilder()
+                .setLastExpiringCertNotificationSentDate(fakeClock.nowUtc())
+                .build()
+                .getLastExpiringCertNotificationSentDate())
+        .isEqualTo(fakeClock.nowUtc());
+  }
+
+  @TestOfyAndSql
+  void testFailure_setLastExpiringCertNotificationSentDate_nullDate() {
+    IllegalArgumentException thrown =
+        assertThrows(
+            IllegalArgumentException.class,
+            () -> new Registrar.Builder().setLastExpiringCertNotificationSentDate(null).build());
+    assertThat(thrown)
+        .hasMessageThat()
+        .isEqualTo("Registrar lastExpiringCertNotificationSentDate cannot be null");
+  }
+
+  @TestOfyAndSql
+  void testSuccess_setLastExpiringFailoverCertNotificationSentDate() {
+    assertThat(
+            registrar
+                .asBuilder()
+                .setLastExpiringFailoverCertNotificationSentDate(fakeClock.nowUtc())
+                .build()
+                .getLastExpiringFailoverCertNotificationSentDate())
+        .isEqualTo(fakeClock.nowUtc());
+  }
+
+  @TestOfyAndSql
+  void testFailure_setLastExpiringFailoverCertNotificationSentDate_nullDate() {
+    IllegalArgumentException thrown =
+        assertThrows(
+            IllegalArgumentException.class,
+            () ->
+                new Registrar.Builder()
+                    .setLastExpiringFailoverCertNotificationSentDate(null)
+                    .build());
+    assertThat(thrown)
+        .hasMessageThat()
+        .isEqualTo("Registrar lastExpiringFailoverCertNotificationSentDate cannot be null");
+  }
+
   @TestOfyAndSql
   void testSuccess_setAllowedTlds() {
     assertThat(

core/src/test/java/google/registry/model/server/KmsSecretRevisionSqlDaoTest.java

@@ -1,86 +0,0 @@
-// Copyright 2020 The Nomulus Authors. All Rights Reserved.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package google.registry.model.server;
-
-import static com.google.common.truth.Truth.assertThat;
-import static google.registry.model.ImmutableObjectSubject.assertAboutImmutableObjects;
-import static google.registry.persistence.transaction.TransactionManagerFactory.jpaTm;
-
-import google.registry.persistence.transaction.JpaTestRules;
-import google.registry.persistence.transaction.JpaTestRules.JpaIntegrationWithCoverageExtension;
-import google.registry.testing.DatastoreEntityExtension;
-import google.registry.testing.FakeClock;
-import java.util.Optional;
-import org.junit.jupiter.api.Order;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.extension.RegisterExtension;
-
-/** Tests for {@link google.registry.model.server.KmsSecretRevisionSqlDao}. */
-public class KmsSecretRevisionSqlDaoTest {
-
-  private final FakeClock fakeClock = new FakeClock();
-
-  @RegisterExtension
-  @Order(value = 1)
-  DatastoreEntityExtension datastoreEntityExtension = new DatastoreEntityExtension();
-
-  @RegisterExtension
-  JpaIntegrationWithCoverageExtension jpa =
-      new JpaTestRules.Builder().withClock(fakeClock).buildIntegrationWithCoverageExtension();
-
-  @Test
-  void testSaveAndRetrieve() {
-    KmsSecretRevision revision = createRevision();
-    jpaTm().transact(() -> KmsSecretRevisionSqlDao.save(revision));
-    Optional<KmsSecretRevision> fromSql =
-        jpaTm().transact(() -> KmsSecretRevisionSqlDao.getLatestRevision("secretName"));
-    assertThat(fromSql.isPresent()).isTrue();
-    assertAboutImmutableObjects().that(revision).isEqualExceptFields(fromSql.get(), "creationTime");
-  }
-
-  @Test
-  void testMultipleRevisions() {
-    KmsSecretRevision revision = createRevision();
-    jpaTm().transact(() -> KmsSecretRevisionSqlDao.save(revision));
-
-    KmsSecretRevision secondRevision = createRevision();
-    secondRevision.encryptedValue = "someOtherValue";
-    jpaTm().transact(() -> KmsSecretRevisionSqlDao.save(secondRevision));
-
-    Optional<KmsSecretRevision> fromSql =
-        jpaTm().transact(() -> KmsSecretRevisionSqlDao.getLatestRevision("secretName"));
-    assertThat(fromSql.isPresent()).isTrue();
-    assertThat(fromSql.get().getEncryptedValue()).isEqualTo("someOtherValue");
-  }
-
-  @Test
-  void testNonexistent() {
-    KmsSecretRevision revision = createRevision();
-    jpaTm().transact(() -> KmsSecretRevisionSqlDao.save(revision));
-    assertThat(
-            jpaTm()
-                .transact(() -> KmsSecretRevisionSqlDao.getLatestRevision("someOtherSecretName"))
-                .isPresent())
-        .isFalse();
-  }
-
-  private KmsSecretRevision createRevision() {
-    return new KmsSecretRevision.Builder()
-        .setEncryptedValue("encrypted")
-        .setKmsCryptoKeyVersionName("version")
-        .setParent("secretName")
-        .build();
-  }
-}

core/src/test/java/google/registry/model/server/KmsSecretRevisionTest.java

@@ -1,68 +0,0 @@
-// Copyright 2017 The Nomulus Authors. All Rights Reserved.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package google.registry.model.server;
-
-import static com.google.common.truth.Truth.assertThat;
-import static google.registry.persistence.transaction.TransactionManagerFactory.ofyTm;
-import static google.registry.testing.DatabaseHelper.persistResource;
-import static org.junit.jupiter.api.Assertions.assertThrows;
-
-import com.google.common.base.Strings;
-import google.registry.testing.AppEngineExtension;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.extension.RegisterExtension;
-
-/** Unit tests for {@link google.registry.model.server.KmsSecretRevision}. */
-public class KmsSecretRevisionTest {
-
-  @RegisterExtension
-  public final AppEngineExtension appEngine =
-      AppEngineExtension.builder().withDatastoreAndCloudSql().build();
-
-  private KmsSecretRevision secretRevision;
-
-  @BeforeEach
-  void beforeEach() {
-    secretRevision =
-        persistResource(
-            new KmsSecretRevision.Builder()
-                .setKmsCryptoKeyVersionName("foo")
-                .setParent("bar")
-                .setEncryptedValue("blah")
-                .build());
-  }
-
-  @Test
-  void test_setEncryptedValue_tooLong_throwsException() {
-    IllegalArgumentException thrown =
-        assertThrows(
-            IllegalArgumentException.class,
-            () ->
-                secretRevision =
-                    persistResource(
-                        new KmsSecretRevision.Builder()
-                            .setKmsCryptoKeyVersionName("foo")
-                            .setParent("bar")
-                            .setEncryptedValue(Strings.repeat("a", 64 * 1024 * 1024 + 1))
-                            .build()));
-    assertThat(thrown).hasMessageThat().contains("Secret is greater than 67108864 bytes");
-  }
-
-  @Test
-  void testPersistence() {
-    assertThat(ofyTm().loadByEntity(secretRevision)).isEqualTo(secretRevision);
-  }
-}

core/src/test/java/google/registry/model/server/KmsSecretTest.java

@@ -1,52 +0,0 @@
-// Copyright 2017 The Nomulus Authors. All Rights Reserved.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package google.registry.model.server;
-
-import static com.google.common.truth.Truth.assertThat;
-import static google.registry.persistence.transaction.TransactionManagerFactory.ofyTm;
-import static google.registry.testing.DatabaseHelper.persistResource;
-
-import google.registry.testing.AppEngineExtension;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.extension.RegisterExtension;
-
-public class KmsSecretTest {
-
-  @RegisterExtension
-  public final AppEngineExtension appEngine =
-      AppEngineExtension.builder().withDatastoreAndCloudSql().build();
-
-  private KmsSecret secret;
-  private KmsSecretRevision secretRevision;
-
-  @BeforeEach
-  void setUp() {
-    secretRevision =
-        persistResource(
-            new KmsSecretRevision.Builder()
-                .setKmsCryptoKeyVersionName("foo")
-                .setParent("bar")
-                .setEncryptedValue("blah")
-                .build());
-
-    secret = persistResource(KmsSecret.create("someSecret", secretRevision));
-  }
-
-  @Test
-  void testPersistence() {
-    assertThat(ofyTm().loadByEntity(secret)).isEqualTo(secret);
-  }
-}

core/src/test/java/google/registry/persistence/transaction/JpaTransactionManagerExtension.java

@@ -202,11 +202,13 @@ abstract class JpaTransactionManagerExtension implements BeforeEachCallback, Aft
     JpaTransactionManagerImpl txnManager = new JpaTransactionManagerImpl(emf, clock);
     cachedTm = TransactionManagerFactory.jpaTm();
     TransactionManagerFactory.setJpaTm(Suppliers.ofInstance(txnManager));
+    JpaTransactionManagerImpl.setReplaySqlToDatastoreOverrideForTest(false);
   }
 
   @Override
   public void afterEach(ExtensionContext context) {
     TransactionManagerFactory.setJpaTm(Suppliers.ofInstance(cachedTm));
+    JpaTransactionManagerImpl.removeReplaySqlToDsOverrideForTest();
     cachedTm = null;
   }
 

core/src/test/java/google/registry/persistence/transaction/JpaTransactionManagerImplTest.java

@@ -17,7 +17,7 @@ package google.registry.persistence.transaction;
 import static com.google.common.collect.ImmutableList.toImmutableList;
 import static com.google.common.truth.Truth.assertThat;
 import static google.registry.persistence.transaction.TransactionManagerFactory.jpaTm;
-import static google.registry.testing.DatabaseHelper.assertDetached;
+import static google.registry.testing.DatabaseHelper.assertDetachedFromEntityManager;
 import static google.registry.testing.TestDataHelper.fileClassPath;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.mockito.ArgumentMatchers.any;
@@ -388,7 +388,8 @@ class JpaTransactionManagerImplTest {
   void load_succeeds() {
     assertThat(jpaTm().transact(() -> jpaTm().exists(theEntity))).isFalse();
     jpaTm().transact(() -> jpaTm().insert(theEntity));
-    TestEntity persisted = jpaTm().transact(() -> assertDetached(jpaTm().loadByKey(theEntityKey)));
+    TestEntity persisted =
+        jpaTm().transact(() -> assertDetachedFromEntityManager(jpaTm().loadByKey(theEntityKey)));
     assertThat(persisted.name).isEqualTo("theEntity");
     assertThat(persisted.data).isEqualTo("foo");
   }
@@ -404,7 +405,8 @@ class JpaTransactionManagerImplTest {
   @Test
   void loadByEntity_succeeds() {
     jpaTm().transact(() -> jpaTm().insert(theEntity));
-    TestEntity persisted = jpaTm().transact(() -> assertDetached(jpaTm().loadByEntity(theEntity)));
+    TestEntity persisted =
+        jpaTm().transact(() -> assertDetachedFromEntityManager(jpaTm().loadByEntity(theEntity)));
     assertThat(persisted.name).isEqualTo("theEntity");
     assertThat(persisted.data).isEqualTo("foo");
   }
@@ -414,7 +416,11 @@ class JpaTransactionManagerImplTest {
     assertThat(jpaTm().transact(() -> jpaTm().exists(theEntity))).isFalse();
     jpaTm().transact(() -> jpaTm().insert(theEntity));
     TestEntity persisted =
-        jpaTm().transact(() -> assertDetached(jpaTm().loadByKeyIfPresent(theEntityKey).get()));
+        jpaTm()
+            .transact(
+                () ->
+                    assertDetachedFromEntityManager(
+                        jpaTm().loadByKeyIfPresent(theEntityKey).get()));
     assertThat(persisted.name).isEqualTo("theEntity");
     assertThat(persisted.data).isEqualTo("foo");
   }
@@ -431,7 +437,9 @@ class JpaTransactionManagerImplTest {
     assertThat(jpaTm().transact(() -> jpaTm().exists(compoundIdEntity))).isFalse();
     jpaTm().transact(() -> jpaTm().insert(compoundIdEntity));
     TestCompoundIdEntity persisted =
-        jpaTm().transact(() -> assertDetached(jpaTm().loadByKey(compoundIdEntityKey)));
+        jpaTm()
+            .transact(
+                () -> assertDetachedFromEntityManager(jpaTm().loadByKey(compoundIdEntityKey)));
     assertThat(persisted.name).isEqualTo("compoundIdEntity");
     assertThat(persisted.age).isEqualTo(10);
     assertThat(persisted.data).isEqualTo("foo");
@@ -450,7 +458,7 @@ class JpaTransactionManagerImplTest {
                               theEntityKey, VKey.createSql(TestEntity.class, "does-not-exist")));
 
               assertThat(results).containsExactly(theEntityKey, theEntity);
-              assertDetached(results.get(theEntityKey));
+              assertDetachedFromEntityManager(results.get(theEntityKey));
             });
   }
 
@@ -463,7 +471,7 @@ class JpaTransactionManagerImplTest {
               ImmutableMap<VKey<? extends TestEntity>, TestEntity> results =
                   jpaTm().loadByKeysIfPresent(ImmutableList.of(theEntityKey));
               assertThat(results).containsExactly(theEntityKey, theEntity);
-              assertDetached(results.get(theEntityKey));
+              assertDetachedFromEntityManager(results.get(theEntityKey));
             });
   }
 
@@ -478,7 +486,7 @@ class JpaTransactionManagerImplTest {
                       .loadByEntitiesIfPresent(
                           ImmutableList.of(theEntity, new TestEntity("does-not-exist", "bar")));
               assertThat(results).containsExactly(theEntity);
-              assertDetached(results.get(0));
+              assertDetachedFromEntityManager(results.get(0));
             });
   }
 
@@ -491,7 +499,7 @@ class JpaTransactionManagerImplTest {
               ImmutableList<TestEntity> results =
                   jpaTm().loadByEntities(ImmutableList.of(theEntity));
               assertThat(results).containsExactly(theEntity);
-              assertDetached(results.get(0));
+              assertDetachedFromEntityManager(results.get(0));
             });
   }
 
@@ -503,7 +511,7 @@ class JpaTransactionManagerImplTest {
             .transact(
                 () ->
                     jpaTm().loadAllOf(TestEntity.class).stream()
-                        .map(DatabaseHelper::assertDetached)
+                        .map(DatabaseHelper::assertDetachedFromEntityManager)
                         .collect(toImmutableList()));
     assertThat(persisted).containsExactlyElementsIn(moreEntities);
   }

core/src/test/java/google/registry/persistence/transaction/QueryComposerTest.java

@@ -77,7 +77,7 @@ public class QueryComposerTest {
                     tm().createQueryComposer(TestEntity.class)
                         .where("name", Comparator.GT, "bravo")
                         .first()
-                        .map(DatabaseHelper::assertDetached)
+                        .map(QueryComposerTest::assertDetachedIfJpa)
                         .get()))
         .isEqualTo(charlie);
     assertThat(
@@ -86,7 +86,7 @@ public class QueryComposerTest {
                     tm().createQueryComposer(TestEntity.class)
                         .where("name", Comparator.GTE, "charlie")
                         .first()
-                        .map(DatabaseHelper::assertDetached)
+                        .map(QueryComposerTest::assertDetachedIfJpa)
                         .get()))
         .isEqualTo(charlie);
     assertThat(
@@ -95,7 +95,7 @@ public class QueryComposerTest {
                     tm().createQueryComposer(TestEntity.class)
                         .where("name", Comparator.LT, "bravo")
                         .first()
-                        .map(DatabaseHelper::assertDetached)
+                        .map(QueryComposerTest::assertDetachedIfJpa)
                         .get()))
         .isEqualTo(alpha);
     assertThat(
@@ -104,7 +104,7 @@ public class QueryComposerTest {
                     tm().createQueryComposer(TestEntity.class)
                         .where("name", Comparator.LTE, "alpha")
                         .first()
-                        .map(DatabaseHelper::assertDetached)
+                        .map(QueryComposerTest::assertDetachedIfJpa)
                         .get()))
         .isEqualTo(alpha);
   }
@@ -125,7 +125,7 @@ public class QueryComposerTest {
     assertThat(
             transactIfJpaTm(
                 () ->
-                    DatabaseHelper.assertDetached(
+                    QueryComposerTest.assertDetachedIfJpa(
                         tm().createQueryComposer(TestEntity.class)
                             .where("name", Comparator.EQ, "alpha")
                             .getSingleResult())))
@@ -174,7 +174,7 @@ public class QueryComposerTest {
                         .createQueryComposer(TestEntity.class)
                         .where("name", Comparator.GT, "alpha")
                         .stream()
-                        .map(DatabaseHelper::assertDetached)
+                        .map(QueryComposerTest::assertDetachedIfJpa)
                         .collect(toImmutableList())))
         .containsExactly(bravo, charlie);
     assertThat(
@@ -184,7 +184,7 @@ public class QueryComposerTest {
                         .createQueryComposer(TestEntity.class)
                         .where("name", Comparator.GTE, "bravo")
                         .stream()
-                        .map(DatabaseHelper::assertDetached)
+                        .map(QueryComposerTest::assertDetachedIfJpa)
                         .collect(toImmutableList())))
         .containsExactly(bravo, charlie);
     assertThat(
@@ -194,7 +194,7 @@ public class QueryComposerTest {
                         .createQueryComposer(TestEntity.class)
                         .where("name", Comparator.LT, "charlie")
                         .stream()
-                        .map(DatabaseHelper::assertDetached)
+                        .map(QueryComposerTest::assertDetachedIfJpa)
                         .collect(toImmutableList())))
         .containsExactly(alpha, bravo);
     assertThat(
@@ -204,7 +204,7 @@ public class QueryComposerTest {
                         .createQueryComposer(TestEntity.class)
                         .where("name", Comparator.LTE, "bravo")
                         .stream()
-                        .map(DatabaseHelper::assertDetached)
+                        .map(QueryComposerTest::assertDetachedIfJpa)
                         .collect(toImmutableList())))
         .containsExactly(alpha, bravo);
   }
@@ -228,7 +228,7 @@ public class QueryComposerTest {
                     tm().createQueryComposer(TestEntity.class)
                         .where("val", Comparator.EQ, 2)
                         .first()
-                        .map(DatabaseHelper::assertDetached)
+                        .map(QueryComposerTest::assertDetachedIfJpa)
                         .get()))
         .isEqualTo(bravo);
   }
@@ -243,7 +243,7 @@ public class QueryComposerTest {
                         .where("val", Comparator.GT, 1)
                         .orderBy("val")
                         .stream()
-                        .map(DatabaseHelper::assertDetached)
+                        .map(QueryComposerTest::assertDetachedIfJpa)
                         .collect(toImmutableList())))
         .containsExactly(bravo, alpha);
   }
@@ -354,6 +354,13 @@ public class QueryComposerTest {
         .contains("The LIKE operation is not supported on Datastore.");
   }
 
+  private static <T> T assertDetachedIfJpa(T entity) {
+    if (!tm().isOfy()) {
+      return DatabaseHelper.assertDetachedFromEntityManager(entity);
+    }
+    return entity;
+  }
+
   @javax.persistence.Entity
   @Entity(name = "QueryComposerTestEntity")
   private static class TestEntity extends ImmutableObject {

core/src/test/java/google/registry/persistence/transaction/TransactionTest.java

@@ -22,36 +22,53 @@ import static org.junit.jupiter.api.Assertions.assertThrows;
 import com.googlecode.objectify.Key;
 import com.googlecode.objectify.annotation.Entity;
 import com.googlecode.objectify.annotation.Id;
-import google.registry.config.RegistryConfig;
 import google.registry.model.ImmutableObject;
+import google.registry.model.ofy.Ofy;
 import google.registry.persistence.VKey;
 import google.registry.testing.AppEngineExtension;
+import google.registry.testing.DatabaseHelper;
+import google.registry.testing.FakeClock;
+import google.registry.testing.InjectExtension;
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.io.ObjectOutputStream;
 import java.io.StreamCorruptedException;
+import org.joda.time.DateTime;
+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;
 
 class TransactionTest {
 
+  private final FakeClock fakeClock = new FakeClock(DateTime.parse("2000-01-01TZ"));
+
   @RegisterExtension
   final AppEngineExtension appEngine =
       AppEngineExtension.builder()
           .withDatastoreAndCloudSql()
+          .withClock(fakeClock)
           .withOfyTestEntities(TestEntity.class)
           .withJpaUnitTestEntities(TestEntity.class)
           .build();
 
+  @RegisterExtension public final InjectExtension inject = new InjectExtension();
+
   private TestEntity fooEntity, barEntity;
 
   @BeforeEach
   void beforeEach() {
+    JpaTransactionManagerImpl.removeReplaySqlToDsOverrideForTest();
+    inject.setStaticField(Ofy.class, "clock", fakeClock);
     fooEntity = new TestEntity("foo");
     barEntity = new TestEntity("bar");
   }
 
+  @AfterEach
+  void afterEach() {
+    DatabaseHelper.removeDatabaseMigrationSchedule();
+  }
+
   @Test
   void testTransactionReplay() {
     Transaction txn = new Transaction.Builder().addUpdate(fooEntity).addUpdate(barEntity).build();
@@ -102,14 +119,13 @@ class TransactionTest {
 
   @Test
   void testTransactionSerialization() throws IOException {
-    RegistryConfig.overrideCloudSqlReplicateTransactions(true);
-    try {
-      jpaTm()
-          .transact(
-              () -> {
-                jpaTm().insert(fooEntity);
-                jpaTm().insert(barEntity);
-              });
+    DatabaseHelper.setMigrationScheduleToSqlPrimary(fakeClock);
+    jpaTm()
+        .transact(
+            () -> {
+              jpaTm().insert(fooEntity);
+              jpaTm().insert(barEntity);
+            });
       TransactionEntity txnEnt =
           jpaTm().transact(() -> jpaTm().loadByKey(VKey.createSql(TransactionEntity.class, 1L)));
       Transaction txn = Transaction.deserialize(txnEnt.getContents());
@@ -125,9 +141,6 @@ class TransactionTest {
       assertThat(
               jpaTm().transact(() -> jpaTm().exists(VKey.createSql(TransactionEntity.class, 2L))))
           .isFalse();
-    } finally {
-      RegistryConfig.overrideCloudSqlReplicateTransactions(false);
-    }
   }
 
   @Test

core/src/test/java/google/registry/schema/integration/SqlIntegrationTestSuite.java

@@ -30,7 +30,6 @@ import google.registry.model.registry.RegistryLockDaoTest;
 import google.registry.model.registry.RegistryTest;
 import google.registry.model.registry.label.ReservedListDaoTest;
 import google.registry.model.reporting.Spec11ThreatMatchTest;
-import google.registry.model.server.KmsSecretRevisionSqlDaoTest;
 import google.registry.model.server.LockTest;
 import google.registry.model.server.ServerSecretTest;
 import google.registry.model.smd.SignedMarkRevocationListDaoTest;
@@ -89,7 +88,6 @@ import org.junit.runner.RunWith;
   DomainBaseSqlTest.class,
   DomainHistoryTest.class,
   HostHistoryTest.class,
-  KmsSecretRevisionSqlDaoTest.class,
   LockTest.class,
   PollMessageTest.class,
   PremiumListDaoTest.class,

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

@@ -26,15 +26,16 @@ import com.google.common.testing.TestLogHandler;
 import com.googlecode.objectify.Key;
 import com.googlecode.objectify.annotation.Entity;
 import com.googlecode.objectify.annotation.Id;
-import google.registry.config.RegistryConfig;
 import google.registry.model.ImmutableObject;
 import google.registry.model.common.DatabaseMigrationStateSchedule;
 import google.registry.model.common.DatabaseMigrationStateSchedule.MigrationState;
 import google.registry.model.ofy.CommitLogBucket;
 import google.registry.model.ofy.Ofy;
 import google.registry.persistence.VKey;
+import google.registry.persistence.transaction.JpaTransactionManagerImpl;
 import google.registry.persistence.transaction.TransactionEntity;
 import google.registry.testing.AppEngineExtension;
+import google.registry.testing.DatabaseHelper;
 import google.registry.testing.FakeClock;
 import google.registry.testing.InjectExtension;
 import java.util.List;
@@ -68,41 +69,21 @@ public class ReplicateToDatastoreActionTest {
 
   @BeforeEach
   public void setUp() {
+    JpaTransactionManagerImpl.removeReplaySqlToDsOverrideForTest();
     injectExtension.setStaticField(Ofy.class, "clock", fakeClock);
     // Use a single bucket to expose timestamp inversion problems.
     injectExtension.setStaticField(
         CommitLogBucket.class, "bucketIdSupplier", Suppliers.ofInstance(1));
     fakeClock.setAutoIncrementByOneMilli();
-    RegistryConfig.overrideCloudSqlReplicateTransactions(true);
+    DatabaseHelper.setMigrationScheduleToSqlPrimary(fakeClock);
     Logger.getLogger(ReplicateToDatastoreAction.class.getCanonicalName()).addHandler(logHandler);
-    DateTime now = fakeClock.nowUtc();
-    ofyTm()
-        .transact(
-            () ->
-                DatabaseMigrationStateSchedule.set(
-                    ImmutableSortedMap.of(
-                        START_OF_TIME,
-                        MigrationState.DATASTORE_ONLY,
-                        now,
-                        MigrationState.DATASTORE_PRIMARY,
-                        now.plusHours(1),
-                        MigrationState.DATASTORE_PRIMARY_READ_ONLY,
-                        now.plusHours(2),
-                        MigrationState.SQL_PRIMARY)));
     fakeClock.advanceBy(Duration.standardDays(1));
   }
 
   @AfterEach
   public void tearDown() {
+    DatabaseHelper.removeDatabaseMigrationSchedule();
     fakeClock.disableAutoIncrement();
-    RegistryConfig.overrideCloudSqlReplicateTransactions(false);
-    ofyTm()
-        .transact(
-            () ->
-                ofyTm()
-                    .loadSingleton(DatabaseMigrationStateSchedule.class)
-                    .ifPresent(ofyTm()::delete));
-    DatabaseMigrationStateSchedule.CACHE.invalidateAll();
   }
 
   @RetryingTest(4)

core/src/test/java/google/registry/testing/DatabaseHelper.java

@@ -35,6 +35,7 @@ import static google.registry.model.ResourceTransferUtils.createTransferResponse
 import static google.registry.model.ofy.ObjectifyService.auditedOfy;
 import static google.registry.model.registry.Registry.TldState.GENERAL_AVAILABILITY;
 import static google.registry.persistence.transaction.TransactionManagerFactory.jpaTm;
+import static google.registry.persistence.transaction.TransactionManagerFactory.ofyTm;
 import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
 import static google.registry.persistence.transaction.TransactionManagerUtil.ofyTmOrDoNothing;
 import static google.registry.persistence.transaction.TransactionManagerUtil.transactIfJpaTm;
@@ -71,6 +72,8 @@ import google.registry.model.EppResourceUtils;
 import google.registry.model.billing.BillingEvent;
 import google.registry.model.billing.BillingEvent.Flag;
 import google.registry.model.billing.BillingEvent.Reason;
+import google.registry.model.common.DatabaseMigrationStateSchedule;
+import google.registry.model.common.DatabaseMigrationStateSchedule.MigrationState;
 import google.registry.model.contact.ContactAuthInfo;
 import google.registry.model.contact.ContactHistory;
 import google.registry.model.contact.ContactResource;
@@ -123,6 +126,7 @@ import org.joda.money.CurrencyUnit;
 import org.joda.money.Money;
 import org.joda.time.DateTime;
 import org.joda.time.DateTimeZone;
+import org.joda.time.Duration;
 
 /** Static utils for setting up test resources. */
 public class DatabaseHelper {
@@ -1351,16 +1355,54 @@ public class DatabaseHelper {
   }
 
   /**
-   * In JPA mode, assert that the given entity is detached from the current entity manager.
+   * Asserts that the given entity is detached from the current JPA entity manager.
    *
    * <p>Returns the original entity object.
    */
-  public static <T> T assertDetached(T entity) {
-    if (!tm().isOfy()) {
-      assertThat(jpaTm().getEntityManager().contains(entity)).isFalse();
-    }
+  public static <T> T assertDetachedFromEntityManager(T entity) {
+    assertThat(jpaTm().getEntityManager().contains(entity)).isFalse();
     return entity;
   }
 
+  /**
+   * Sets a SQL_PRIMARY state on the {@link DatabaseMigrationStateSchedule}.
+   *
+   * <p>In order to allow for tests to manipulate the clock how they need, we start the transitions
+   * one millisecond after the clock's current time (in case the clock's current value is
+   * START_OF_TIME). We then advance the clock one second so that we're in the SQL_PRIMARY phase.
+   *
+   * <p>We must use the current time, otherwise the setting of the migration state will fail due to
+   * an invalid transition.
+   */
+  public static void setMigrationScheduleToSqlPrimary(FakeClock fakeClock) {
+    DateTime now = fakeClock.nowUtc();
+    ofyTm()
+        .transact(
+            () ->
+                DatabaseMigrationStateSchedule.set(
+                    ImmutableSortedMap.of(
+                        START_OF_TIME,
+                        MigrationState.DATASTORE_ONLY,
+                        now.plusMillis(1),
+                        MigrationState.DATASTORE_PRIMARY,
+                        now.plusMillis(2),
+                        MigrationState.DATASTORE_PRIMARY_READ_ONLY,
+                        now.plusMillis(3),
+                        MigrationState.SQL_PRIMARY)));
+    fakeClock.advanceBy(Duration.standardSeconds(1));
+  }
+
+  /** Removes the database migration schedule, in essence transitioning to DATASTORE_ONLY. */
+  public static void removeDatabaseMigrationSchedule() {
+    // use the raw calls because going SQL_PRIMARY -> DATASTORE_ONLY is not valid
+    ofyTm()
+        .transact(
+            () ->
+                ofyTm()
+                    .loadSingleton(DatabaseMigrationStateSchedule.class)
+                    .ifPresent(ofyTm()::delete));
+    DatabaseMigrationStateSchedule.CACHE.invalidateAll();
+  }
+
   private DatabaseHelper() {}
 }

core/src/test/java/google/registry/testing/ReplayExtension.java

@@ -22,12 +22,12 @@ 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.JpaTransactionManagerImpl;
 import google.registry.persistence.transaction.TransactionEntity;
 import google.registry.schema.replay.DatastoreEntity;
 import google.registry.schema.replay.ReplicateToDatastoreAction;
@@ -92,7 +92,7 @@ public class ReplayExtension implements BeforeEachCallback, AfterEachCallback {
     // transaction manager gets injected.
     inOfyContext = DualDatabaseTestInvocationContextProvider.inOfyContext(context);
     if (sqlToDsReplicator != null && !inOfyContext) {
-      RegistryConfig.overrideCloudSqlReplicateTransactions(true);
+      JpaTransactionManagerImpl.setReplaySqlToDatastoreOverrideForTest(true);
     }
 
     context.getStore(ExtensionContext.Namespace.GLOBAL).put(ReplayExtension.class, this);
@@ -105,7 +105,7 @@ public class ReplayExtension implements BeforeEachCallback, AfterEachCallback {
     replay();
     injectExtension.afterEach(context);
     if (sqlToDsReplicator != null) {
-      RegistryConfig.overrideCloudSqlReplicateTransactions(false);
+      JpaTransactionManagerImpl.setReplaySqlToDatastoreOverrideForTest(false);
     }
   }
 
@@ -116,7 +116,6 @@ public class ReplayExtension implements BeforeEachCallback, AfterEachCallback {
           "PremiumListEntry",
           "ReservedList",
           "RdeRevision",
-          "KmsSecretRevision",
           "ServerSecret",
           "SignedMarkRevocationList",
           "ClaimsListShard",

core/src/test/java/google/registry/testing/TestObject.java

@@ -75,7 +75,8 @@ public class TestObject extends ImmutableObject implements DatastoreAndSqlEntity
     beforeSqlDeleteCallCount++;
   }
 
-  public static void beforeSqlSave(TestObject testObject) {
+  @Override
+  public void beforeSqlSaveOnReplay() {
     beforeSqlSaveCallCount++;
   }
 

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

@@ -67,6 +67,7 @@ class CreateRegistrarCommandTest extends CommandTestCase<CreateRegistrarCommand>
         new CertificateChecker(
             ImmutableSortedMap.of(START_OF_TIME, 825, DateTime.parse("2020-09-01T00:00:00Z"), 398),
             30,
+            15,
             2048,
             ImmutableSet.of("secp256r1", "secp384r1"),
             fakeClock);

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

@@ -21,6 +21,7 @@ import static google.registry.util.DateTimeUtils.START_OF_TIME;
 import com.google.common.collect.ImmutableSortedMap;
 import google.registry.model.common.DatabaseMigrationStateSchedule;
 import google.registry.model.common.DatabaseMigrationStateSchedule.MigrationState;
+import google.registry.testing.DatabaseHelper;
 import google.registry.testing.DualDatabaseTest;
 import google.registry.testing.TestOfyAndSql;
 import org.joda.time.DateTime;
@@ -33,7 +34,7 @@ public class GetDatabaseMigrationStateCommandTest
 
   @AfterEach
   void afterEach() {
-    ofyTm().transact(() -> DatabaseMigrationStateSchedule.set(DEFAULT_TRANSITION_MAP.toValueMap()));
+    DatabaseHelper.removeDatabaseMigrationSchedule();
   }
 
   @TestOfyAndSql

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

@@ -25,32 +25,20 @@ import com.beust.jcommander.ParameterException;
 import com.google.common.collect.ImmutableSortedMap;
 import google.registry.model.common.DatabaseMigrationStateSchedule;
 import google.registry.model.common.DatabaseMigrationStateSchedule.MigrationState;
+import google.registry.testing.DatabaseHelper;
 import google.registry.testing.DualDatabaseTest;
 import google.registry.testing.TestOfyAndSql;
 import org.joda.time.DateTime;
 import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.BeforeEach;
 
 /** Tests for {@link SetDatabaseMigrationStateCommand}. */
 @DualDatabaseTest
 public class SetDatabaseMigrationStateCommandTest
     extends CommandTestCase<SetDatabaseMigrationStateCommand> {
 
-  @BeforeEach
-  void beforeEach() {
-    // clear out any static state that may have been persisted
-    ofyTm()
-        .transact(
-            () ->
-                ofyTm()
-                    .loadSingleton(DatabaseMigrationStateSchedule.class)
-                    .ifPresent(ofyTm()::delete));
-    DatabaseMigrationStateSchedule.CACHE.invalidateAll();
-  }
-
   @AfterEach
   void afterEach() {
-    ofyTm().transact(() -> DatabaseMigrationStateSchedule.set(DEFAULT_TRANSITION_MAP.toValueMap()));
+    DatabaseHelper.removeDatabaseMigrationSchedule();
   }
 
   @TestOfyAndSql

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

@@ -53,6 +53,7 @@ class UpdateRegistrarCommandTest extends CommandTestCase<UpdateRegistrarCommand>
         new CertificateChecker(
             ImmutableSortedMap.of(START_OF_TIME, 825, DateTime.parse("2020-09-01T00:00:00Z"), 398),
             30,
+            15,
             2048,
             ImmutableSet.of("secp256r1", "secp384r1"),
             fakeClock);

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

@@ -65,6 +65,7 @@ class ValidateLoginCredentialsCommandTest extends CommandTestCase<ValidateLoginC
         new CertificateChecker(
             ImmutableSortedMap.of(START_OF_TIME, 825, DateTime.parse("2020-09-01T00:00:00Z"), 398),
             30,
+            15,
             2048,
             ImmutableSet.of("secp256r1", "secp384r1"),
             fakeClock);

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

@@ -128,6 +128,7 @@ public abstract class RegistrarSettingsActionTestCase {
         new CertificateChecker(
             ImmutableSortedMap.of(START_OF_TIME, 825, DateTime.parse("2020-09-01T00:00:00Z"), 398),
             30,
+            15,
             2048,
             ImmutableSet.of("secp256r1", "secp384r1"),
             clock);

core/src/test/resources/google/registry/export/backup_kinds.txt

@@ -11,8 +11,6 @@ ForeignKeyDomainIndex
 ForeignKeyHostIndex
 HistoryEntry
 HostResource
-KmsSecret
-KmsSecretRevision
 LastSqlTransaction
 Modification
 OneTime

core/src/test/resources/google/registry/export/crosstld_kinds.txt

@@ -2,8 +2,6 @@ ClaimsList
 ClaimsListSingleton
 Cursor
 DatabaseMigrationStateSchedule
-KmsSecret
-KmsSecretRevision
 PremiumList
 PremiumListEntry
 PremiumListRevision

core/src/test/resources/google/registry/export/reporting_kinds.txt

@@ -8,8 +8,6 @@ ForeignKeyDomainIndex
 ForeignKeyHostIndex
 HistoryEntry
 HostResource
-KmsSecret
-KmsSecretRevision
 Modification
 OneTime
 PollMessage

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

@@ -600,6 +600,8 @@ class google.registry.model.registrar.Registrar {
   java.util.Set<java.lang.String> allowedTlds;
   java.util.Set<java.lang.String> rdapBaseUrls;
   org.joda.time.DateTime lastCertificateUpdateTime;
+  org.joda.time.DateTime lastExpiringCertNotificationSentDate;
+  org.joda.time.DateTime lastExpiringFailoverCertNotificationSentDate;
 }
 class google.registry.model.registrar.Registrar$BillingAccountEntry {
   java.lang.String accountId;
@@ -825,18 +827,6 @@ enum google.registry.model.reporting.HistoryEntry$Type {
   RDE_IMPORT;
   SYNTHETIC;
 }
-class google.registry.model.server.KmsSecret {
-  @Id java.lang.String name;
-  @Parent com.googlecode.objectify.Key<google.registry.model.common.EntityGroupRoot> parent;
-  com.googlecode.objectify.Key<google.registry.model.server.KmsSecretRevision> latestRevision;
-}
-class google.registry.model.server.KmsSecretRevision {
-  @Id long revisionKey;
-  @Parent com.googlecode.objectify.Key<google.registry.model.server.KmsSecret> parent;
-  google.registry.model.CreateAutoTimestamp creationTime;
-  java.lang.String encryptedValue;
-  java.lang.String kmsCryptoKeyVersionName;
-}
 class google.registry.model.server.Lock {
   @Id java.lang.String lockId;
   java.lang.String requestLogId;

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

@@ -95,3 +95,4 @@ V94__rename_lock_scope.sql
 V95__add_contacts_indexes_on_domain.sql
 V96__rename_sql_checkpoint_fields.sql
 V97__add_recurrence_history_id_column_to_onetime.sql
+V98__add_last_expiring_cert_notification_sent_date.sql

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

@@ -0,0 +1,18 @@
+-- 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.
+
+alter table "Registrar"
+    add column if not exists "last_expiring_cert_notification_sent_date" timestamptz;
+alter table "Registrar"
+    add column if not exists "last_expiring_failover_cert_notification_sent_date" timestamptz;

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

@@ -491,15 +491,6 @@
         primary key (host_repo_id, history_revision_id)
     );
 
-    create table "KmsSecret" (
-       revision_id int8 not null,
-        creation_time timestamptz not null,
-        encrypted_value text not null,
-        crypto_key_version_name text not null,
-        secret_name text not null,
-        primary key (revision_id)
-    );
-
     create table "Lock" (
        resource_name text not null,
         scope text not null,
@@ -590,6 +581,8 @@
         i18n_address_zip text,
         ip_address_allow_list text[],
         last_certificate_update_time timestamptz,
+        last_expiring_cert_notification_sent_date timestamptz,
+        last_expiring_failover_cert_notification_sent_date timestamptz,
         last_update_time timestamptz not null,
         localized_address_city text,
         localized_address_country_code text,
@@ -806,7 +799,6 @@ create index IDX1iy7njgb7wjmj9piml4l2g0qi on "HostHistory" (history_registrar_id
 create index IDXkkwbwcwvrdkkqothkiye4jiff on "HostHistory" (host_name);
 create index IDXknk8gmj7s47q56cwpa6rmpt5l on "HostHistory" (history_type);
 create index IDX67qwkjtlq5q8dv6egtrtnhqi7 on "HostHistory" (history_modification_time);
-create index IDXli9nil3s4t4p21i3xluvvilb7 on "KmsSecret" (secret_name);
 create index IDXe7wu46c7wpvfmfnj4565abibp on "PollMessage" (registrar_id);
 create index IDXaydgox62uno9qx8cjlj5lauye on "PollMessage" (event_time);
 create index premiumlist_name_idx on "PremiumList" (name);

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

@@ -800,7 +800,9 @@ CREATE TABLE public."Registrar" (
     state text,
     type text NOT NULL,
     url text,
-    whois_server text
+    whois_server text,
+    last_expiring_cert_notification_sent_date timestamp with time zone,
+    last_expiring_failover_cert_notification_sent_date timestamp with time zone
 );