Track and replay Transaction table gaps (#1557)

* Track and replay Transaction table gaps

Id gaps in the Transaction table can be the result of a transactions committed
out of order.  To deal with this, keep track of gaps for up to five minutes
and check to see if they've been back-filled prior to applying the next batch
of transactions during reply.

* Changes for review

* Calculate gap expiration time before gap queries

* Reformat.

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

@@ -41,6 +41,7 @@ import google.registry.model.rde.RdeRevision;
 import google.registry.model.registrar.Registrar;
 import google.registry.model.registrar.RegistrarContact;
 import google.registry.model.replay.LastSqlTransaction;
+import google.registry.model.replay.ReplayGap;
 import google.registry.model.reporting.HistoryEntry;
 import google.registry.model.server.Lock;
 import google.registry.model.server.ServerSecret;
@@ -74,6 +75,7 @@ public class ClassPathManagerTest {
     assertThat(ClassPathManager.getClass("HostResource")).isEqualTo(HostResource.class);
     assertThat(ClassPathManager.getClass("Recurring")).isEqualTo(Recurring.class);
     assertThat(ClassPathManager.getClass("Registrar")).isEqualTo(Registrar.class);
+    assertThat(ClassPathManager.getClass("ReplayGap")).isEqualTo(ReplayGap.class);
     assertThat(ClassPathManager.getClass("ContactResource")).isEqualTo(ContactResource.class);
     assertThat(ClassPathManager.getClass("Cancellation")).isEqualTo(Cancellation.class);
     assertThat(ClassPathManager.getClass("RegistrarContact")).isEqualTo(RegistrarContact.class);
@@ -141,6 +143,7 @@ public class ClassPathManagerTest {
     assertThat(ClassPathManager.getClassName(HostResource.class)).isEqualTo("HostResource");
     assertThat(ClassPathManager.getClassName(Recurring.class)).isEqualTo("Recurring");
     assertThat(ClassPathManager.getClassName(Registrar.class)).isEqualTo("Registrar");
+    assertThat(ClassPathManager.getClassName(ReplayGap.class)).isEqualTo("ReplayGap");
     assertThat(ClassPathManager.getClassName(ContactResource.class)).isEqualTo("ContactResource");
     assertThat(ClassPathManager.getClassName(Cancellation.class)).isEqualTo("Cancellation");
     assertThat(ClassPathManager.getClassName(RegistrarContact.class)).isEqualTo("RegistrarContact");

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

@@ -27,11 +27,13 @@ import static org.junit.jupiter.api.Assumptions.assumeTrue;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
+import com.google.common.base.Supplier;
 import com.google.common.base.Suppliers;
 import com.google.common.collect.ImmutableSortedMap;
 import com.google.common.flogger.FluentLogger;
 import com.google.common.testing.TestLogHandler;
 import com.google.common.truth.Truth8;
+import com.googlecode.objectify.Key;
 import google.registry.model.common.DatabaseMigrationStateSchedule;
 import google.registry.model.common.DatabaseMigrationStateSchedule.MigrationState;
 import google.registry.model.domain.token.AllocationToken;
@@ -39,6 +41,8 @@ import google.registry.model.domain.token.AllocationToken.TokenType;
 import google.registry.model.ofy.CommitLogBucket;
 import google.registry.model.ofy.Ofy;
 import google.registry.model.server.Lock;
+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;
@@ -48,9 +52,13 @@ import google.registry.testing.InjectExtension;
 import google.registry.testing.ReplayExtension;
 import google.registry.testing.TestObject;
 import google.registry.util.RequestStatusChecker;
+import java.lang.reflect.Proxy;
 import java.util.List;
 import java.util.logging.Level;
 import java.util.logging.Logger;
+import javax.persistence.EntityManager;
+import javax.persistence.EntityManagerFactory;
+import javax.persistence.EntityTransaction;
 import org.joda.time.DateTime;
 import org.joda.time.Duration;
 import org.junit.jupiter.api.AfterEach;
@@ -241,6 +249,64 @@ public class ReplicateToDatastoreActionTest {
     assertThat(ofyTm().transact(() -> LastSqlTransaction.load()).getTransactionId()).isEqualTo(4);
   }
 
+  @Test
+  void testTransactionGapReplay() {
+    insertInDb(TestObject.create("foo"));
+    DeferredCommit deferred = new DeferredCommit(fakeClock);
+    TestObject bar = TestObject.create("bar");
+    deferred.transact(() -> jpaTm().insert(bar));
+    TestObject baz = TestObject.create("baz");
+    insertInDb(baz);
+
+    action.run();
+    assertThat(ofyTm().transact(() -> ofyTm().loadByKeyIfPresent(bar.key())).isPresent()).isFalse();
+    assertThat(ofyTm().transact(() -> ofyTm().loadByKey(baz.key()))).isEqualTo(baz);
+    VKey<ReplayGap> gapKey = VKey.createOfy(ReplayGap.class, Key.create(ReplayGap.class, 2));
+    Truth8.assertThat(ofyTm().transact(() -> ofyTm().loadByKeyIfPresent(gapKey))).isPresent();
+
+    deferred.commit();
+    resetAction();
+    action.run();
+
+    assertThat(ofyTm().transact(() -> ofyTm().loadByKey(bar.key()))).isEqualTo(bar);
+    // Verify that the gap record has been cleaned up.
+    assertThat(ofyTm().transact(() -> ofyTm().loadByKeyIfPresent(gapKey).isPresent())).isFalse();
+  }
+
+  @Test
+  void testGapRecordExpiration() {
+    insertInDb(TestObject.create("foo"));
+
+    // Fail a transaction, create a gap.
+    try {
+      jpaTm()
+          .transact(
+              () -> {
+                jpaTm().insert(TestObject.create("other"));
+                // Explicitly save the transaction entity to force the id update.
+                jpaTm().insert(new TransactionEntity(new byte[] {1, 2, 3}));
+                throw new RuntimeException("fail!!!");
+              });
+    } catch (Exception e) {
+      logger.atInfo().log("Got expected exception.");
+    }
+
+    insertInDb(TestObject.create("bar"));
+
+    action.run();
+
+    // Verify that the gap record has been created
+    VKey<ReplayGap> gapKey = VKey.createOfy(ReplayGap.class, Key.create(ReplayGap.class, 2));
+    Truth8.assertThat(ofyTm().transact(() -> ofyTm().loadByKeyIfPresent(gapKey))).isPresent();
+
+    fakeClock.advanceBy(Duration.millis(ReplicateToDatastoreAction.MAX_GAP_RETENTION_MILLIS + 1));
+    resetAction();
+    action.run();
+
+    // Verify that the gap record has been destroyed.
+    assertThat(ofyTm().transact(() -> ofyTm().loadByKeyIfPresent(gapKey)).isPresent()).isFalse();
+  }
+
   @Test
   void testBeforeDatastoreSaveCallback() {
     assumeTrue(ReplayExtension.replayTestsEnabled());
@@ -327,4 +393,63 @@ public class ReplicateToDatastoreActionTest {
     when(requestStatusChecker.getLogId()).thenReturn("logId");
     action = new ReplicateToDatastoreAction(fakeClock, requestStatusChecker, response);
   }
+
+  /**
+   * Deep fake of EntityManagerFactory -> EntityManager -> EntityTransaction that allows us to defer
+   * the actual commit until after the other transactions are replayed.
+   */
+  static class DeferredCommit {
+
+    FakeClock clock;
+    EntityTransaction realTransaction;
+
+    DeferredCommit(FakeClock clock) {
+      this.clock = clock;
+    }
+
+    private static <T> T makeProxy(
+        Class<T> iface, T delegate, String method, Supplier<?> supplier) {
+      return (T)
+          Proxy.newProxyInstance(
+              delegate.getClass().getClassLoader(),
+              new Class[] {iface},
+              (proxy, meth, args) -> {
+                if (meth.getName().equals(method)) {
+                  return supplier.get();
+                } else {
+                  return meth.invoke(delegate, args);
+                }
+              });
+    }
+
+    EntityManager createEntityManagerProxy(EntityManager orgEntityManager) {
+      return makeProxy(
+          EntityManager.class,
+          orgEntityManager,
+          "getTransaction",
+          () ->
+              makeProxy(
+                  EntityTransaction.class,
+                  realTransaction = orgEntityManager.getTransaction(),
+                  "commit",
+                  () -> null));
+    }
+
+    void commit() {
+      realTransaction.commit();
+    }
+
+    void transact(Runnable runnable) {
+      EntityManagerFactory orgEmf =
+          jpaTm().transact(() -> jpaTm().getEntityManager().getEntityManagerFactory());
+      EntityManagerFactory emfProxy =
+          makeProxy(
+              EntityManagerFactory.class,
+              orgEmf,
+              "createEntityManager",
+              () -> createEntityManagerProxy(orgEmf.createEntityManager()));
+
+      new JpaTransactionManagerImpl(emfProxy, clock).transact(runnable);
+    }
+  }
 }

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

@@ -641,6 +641,10 @@ class google.registry.model.replay.LastSqlTransaction {
   @Id long id;
   long transactionId;
 }
+class google.registry.model.replay.ReplayGap {
+  @Id long transactionId;
+  org.joda.time.DateTime timestamp;
+}
 class google.registry.model.reporting.DomainTransactionRecord {
   google.registry.model.reporting.DomainTransactionRecord$TransactionReportField reportField;
   java.lang.Integer reportAmount;