Allow replicateToDatastore to skip gaps (#1539)

* Allow replicateToDatastore to skip gaps

As it turns out, gaps in the transaction id sequence number are expected
because rollbacks do not rollback sequence numbers.

To deal with this, stop checking these.

This change is not adequate in and of itself, as it is possible for a gap to
be introduced if two transactions are committed out of order of their sequence
number.  We are currently discussing several strategies to mitigate this.

* Remove println, add a record verification

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

@@ -29,6 +29,8 @@ import google.registry.request.Parameter;
 import google.registry.request.Response;
 import google.registry.request.auth.Auth;
 import google.registry.util.Sleeper;
+import java.io.ByteArrayOutputStream;
+import java.io.PrintStream;
 import java.util.Optional;
 import javax.inject.Inject;
 import org.joda.time.DateTime;
@@ -114,9 +116,22 @@ public class SyncDatastoreToSqlSnapshotAction implements Runnable {
       response.setPayload(
           String.format(SUCCESS_RESPONSE_TEMPLATE, sqlSnapshotId, checkpoint.getCheckpointTime()));
       return;
-    } catch (Exception e) {
+    } catch (Throwable e) {
+      logger.atSevere().withCause(e).log("Failed to sync Datastore to SQL.");
       response.setStatus(SC_INTERNAL_SERVER_ERROR);
-      response.setPayload(e.getMessage());
+      response.setPayload(getStackTrace(e));
+    }
+  }
+
+  private static String getStackTrace(Throwable e) {
+    try {
+      ByteArrayOutputStream bis = new ByteArrayOutputStream();
+      PrintStream printStream = new PrintStream(bis);
+      e.printStackTrace(printStream);
+      printStream.close();
+      return bis.toString();
+    } catch (RuntimeException re) {
+      return re.getMessage();
     }
   }
 

core/src/main/java/google/registry/model/replay/ReplicateToDatastoreAction.java

@@ -128,13 +128,10 @@ public class ReplicateToDatastoreAction implements Runnable {
                 LastSqlTransaction lastSqlTxn = LastSqlTransaction.load();
                 long nextTxnId = lastSqlTxn.getTransactionId() + 1;
                 if (nextTxnId < txnEntity.getId()) {
-                  // We're missing a transaction.  This is bad.  Transaction ids are supposed to
-                  // increase monotonically, so we abort rather than applying anything out of
-                  // order.
-                  throw new IllegalStateException(
-                      String.format(
-                          "Missing transaction: last txn id = %s, next available txn = %s",
-                          nextTxnId - 1, txnEntity.getId()));
+                  // Missing transaction id.  This can happen normally.  If a transaction gets
+                  // rolled back, the sequence counter doesn't.
+                  logger.atWarning().log(
+                      "Ignoring transaction %s, which does not exist.", nextTxnId);
                 } else if (nextTxnId > txnEntity.getId()) {
                   // We've already replayed this transaction.  This shouldn't happen, as GAE cron
                   // is supposed to avoid overruns and this action shouldn't be executed from any

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

@@ -14,6 +14,7 @@
 
 package google.registry.persistence.transaction;
 
+import com.google.common.annotations.VisibleForTesting;
 import google.registry.model.ImmutableObject;
 import google.registry.model.replay.SqlOnlyEntity;
 import javax.persistence.Entity;
@@ -39,7 +40,8 @@ public class TransactionEntity extends ImmutableObject implements SqlOnlyEntity
 
   TransactionEntity() {}
 
-  TransactionEntity(byte[] contents) {
+  @VisibleForTesting
+  public TransactionEntity(byte[] contents) {
     this.contents = contents;
   }
 

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

@@ -21,16 +21,15 @@ import static google.registry.persistence.transaction.TransactionManagerFactory.
 import static google.registry.testing.DatabaseHelper.insertInDb;
 import static google.registry.testing.LogsSubject.assertAboutLogs;
 import static google.registry.util.DateTimeUtils.START_OF_TIME;
-import static javax.servlet.http.HttpServletResponse.SC_INTERNAL_SERVER_ERROR;
 import static javax.servlet.http.HttpServletResponse.SC_NO_CONTENT;
 import static javax.servlet.http.HttpServletResponse.SC_OK;
-import static org.junit.Assert.assertThrows;
 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.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 google.registry.model.common.DatabaseMigrationStateSchedule;
@@ -63,6 +62,7 @@ import org.junitpioneer.jupiter.RetryingTest;
 public class ReplicateToDatastoreActionTest {
 
   private final FakeClock fakeClock = new FakeClock(DateTime.parse("2000-01-01TZ"));
+  private static final FluentLogger logger = FluentLogger.forEnclosingClass();
 
   @RegisterExtension
   public final AppEngineExtension appEngine =
@@ -203,41 +203,38 @@ public class ReplicateToDatastoreActionTest {
   }
 
   @RetryingTest(4)
-  void testMissingTransactions() {
-    assumeTrue(ReplayExtension.replayTestsEnabled());
-
-    // Write a transaction (should have a transaction id of 1).
+  void testNoTransactionIdUpdate() {
+    // Create an object.
     TestObject foo = TestObject.create("foo");
     insertInDb(foo);
 
-    // Force the last transaction id back to -1 so that we look for transaction 0.
-    ofyTm().transact(() -> ofyTm().insert(new LastSqlTransaction(-1)));
+    // Fail during the transaction to delete it.
+    try {
+      jpaTm()
+          .transact(
+              () -> {
+                jpaTm().delete(foo.key());
+                // 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.");
+    }
 
+    TestObject bar = TestObject.create("bar");
+    insertInDb(bar);
+
+    // Make sure we have only the expected transaction ids.
     List<TransactionEntity> txns = action.getTransactionBatchAtSnapshot();
-    assertThat(txns).hasSize(1);
-    assertThat(assertThrows(IllegalStateException.class, () -> applyTransaction(txns.get(0))))
-        .hasMessageThat()
-        .isEqualTo("Missing transaction: last txn id = -1, next available txn = 1");
-  }
+    assertThat(txns).hasSize(2);
+    for (TransactionEntity txn : txns) {
+      assertThat(txn.getId()).isNotEqualTo(2);
+      applyTransaction(txn);
+    }
 
-  @Test
-  void testMissingTransactions_fullTask() {
-    assumeTrue(ReplayExtension.replayTestsEnabled());
-
-    // Write a transaction (should have a transaction id of 1).
-    TestObject foo = TestObject.create("foo");
-    insertInDb(foo);
-
-    // Force the last transaction id back to -1 so that we look for transaction 0.
-    ofyTm().transact(() -> ofyTm().insert(new LastSqlTransaction(-1)));
-    action.run();
-    assertAboutLogs()
-        .that(logHandler)
-        .hasSevereLogWithCause(
-            new IllegalStateException(
-                "Missing transaction: last txn id = -1, next available txn = 1"));
-    assertThat(response.getStatus()).isEqualTo(SC_INTERNAL_SERVER_ERROR);
-    assertThat(response.getPayload()).isEqualTo("Errored out replaying files.");
+    assertThat(ofyTm().transact(() -> ofyTm().loadByKey(foo.key()))).isEqualTo(foo);
+    assertThat(ofyTm().transact(() -> ofyTm().loadByKey(bar.key()))).isEqualTo(bar);
   }
 
   @Test

core/src/test/java/google/registry/webdriver/TestServerExtension.java

@@ -33,6 +33,7 @@ import google.registry.testing.UserInfo;
 import java.net.URL;
 import java.net.UnknownHostException;
 import java.nio.file.Path;
+import java.time.Duration;
 import java.util.Optional;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.Callable;
@@ -53,6 +54,8 @@ import org.junit.jupiter.api.extension.ExtensionContext;
  */
 public final class TestServerExtension implements BeforeEachCallback, AfterEachCallback {
 
+  private static final Duration SERVER_STATUS_POLLING_INTERVAL = Duration.ofSeconds(5);
+
   private final ImmutableList<Fixture> fixtures;
   private final AppEngineExtension appEngineExtension;
   private final BlockingQueue<FutureTask<?>> jobs = new LinkedBlockingDeque<>();
@@ -107,8 +110,11 @@ public final class TestServerExtension implements BeforeEachCallback, AfterEachC
     serverThread = new Thread(server);
     synchronized (this) {
       serverThread.start();
-      while (!server.isRunning) {
-        this.wait();
+      while (server.serverStatus.equals(ServerStatus.NOT_STARTED)) {
+        this.wait(SERVER_STATUS_POLLING_INTERVAL.toMillis());
+      }
+      if (server.serverStatus.equals(ServerStatus.FAILED)) {
+        throw new RuntimeException("TestServer failed to start. See log for details.");
       }
     }
   }
@@ -163,6 +169,12 @@ public final class TestServerExtension implements BeforeEachCallback, AfterEachC
     return job.get();
   }
 
+  enum ServerStatus {
+    NOT_STARTED,
+    RUNNING,
+    FAILED
+  }
+
   private final class Server implements Runnable {
 
     private ExtensionContext context;
@@ -171,7 +183,7 @@ public final class TestServerExtension implements BeforeEachCallback, AfterEachC
       this.context = context;
     }
 
-    private volatile boolean isRunning = false;
+    private volatile ServerStatus serverStatus = ServerStatus.NOT_STARTED;
 
     @Override
     public void run() {
@@ -185,6 +197,7 @@ public final class TestServerExtension implements BeforeEachCallback, AfterEachC
           appEngineExtension.afterEach(context);
         }
       } catch (Throwable e) {
+        serverStatus = ServerStatus.FAILED;
         throw new RuntimeException(e);
       }
     }
@@ -196,7 +209,7 @@ public final class TestServerExtension implements BeforeEachCallback, AfterEachC
       testServer.start();
       System.out.printf("TestServerExtension is listening on: %s\n", testServer.getUrl("/"));
       synchronized (TestServerExtension.this) {
-        isRunning = true;
+        serverStatus = ServerStatus.RUNNING;
         TestServerExtension.this.notify();
       }
       try {