Fix issues with saving and deleting gap records (#1561)

* Fix issues with saving and deleting gap records

Datastore limits us to mutating up to 25 records per transaction.  We
sometimes exceed that when deleting expired gap records.  In addition, it is
theoretically possible for us to accumulate enough continuous gap records to
exceed this count while replaying the original transaction.

Deal with deletion by breaking up the gap records to be deleted into a batch
size that is small enough to be deleted transactionally (in practice, we don't
much care about the transactionality but it doesn't seem like we can delete
batches without it).

Deal with the possibility of too many additions by always breaking out gap
record storage and last transaction number updates into their own
transaction(s) (separate from the replay of the original SQL transaction).

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

@@ -273,6 +273,41 @@ public class ReplicateToDatastoreActionTest {
     assertThat(ofyTm().transact(() -> ofyTm().loadByKeyIfPresent(gapKey).isPresent())).isFalse();
   }
 
+  /** Verify that we can handle creation and deletion of > 25 gap records. */
+  @Test
+  void testLargeNumberOfGaps() {
+    // Fail thirty transactions.
+    for (int i = 0; i < 30; ++i) {
+      try {
+        jpaTm()
+            .transact(
+                () -> {
+                  insertInDb(TestObject.create("foo"));
+                  // 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) {
+        ;
+      }
+    }
+
+    TestObject bar = TestObject.create("bar");
+    insertInDb(bar);
+
+    // Verify that the transaction was successfully applied and that we have generated 30 gap
+    // records.
+    action.run();
+    Truth8.assertThat(ofyTm().transact(() -> ofyTm().loadByKeyIfPresent(bar.key()))).isPresent();
+    assertThat(ofyTm().loadAllOf(ReplayGap.class).size()).isEqualTo(30);
+
+    // Verify that we can clean up this many gap records after expiration.
+    fakeClock.advanceBy(Duration.millis(ReplicateToDatastoreAction.MAX_GAP_RETENTION_MILLIS + 1));
+    resetAction();
+    action.run();
+    assertThat(ofyTm().loadAllOf(ReplayGap.class).size()).isEqualTo(0);
+  }
+
   @Test
   void testGapRecordExpiration() {
     insertInDb(TestObject.create("foo"));