Use the DB migration schedule for SQL->DS replay (#1242)

This is instead of the current configuration parameter.

In addition, this adds some helpers to DatabaseHelper to make the
transitions easier, since we more frequently need to alter + reset the
schedule.

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

@@ -1494,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). */

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.

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/persistence/transaction/JpaTransactionManagerImpl.java

@@ -30,8 +30,9 @@ 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;
@@ -103,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;
@@ -736,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;
@@ -755,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();
       }
     }