Add dual write for Cursors (#414)

* Add dual write for Cursors

* Fix UpdateCursorCommand to dual write multiple cursors

* Small fixes

* Make UpdateCursorsCommand implement CommandWithCloudSql

core/src/main/java/google/registry/batch/ExpandRecurringBillingEventsAction.java

@@ -58,6 +58,7 @@ import google.registry.request.Action;
 import google.registry.request.Parameter;
 import google.registry.request.Response;
 import google.registry.request.auth.Auth;
+import google.registry.schema.cursor.CursorDao;
 import google.registry.util.Clock;
 import java.util.Optional;
 import java.util.Set;
@@ -313,8 +314,7 @@ public class ExpandRecurringBillingEventsAction implements Runnable {
       logger.atInfo().log(
           "Recurring event expansion %s complete for billing event range [%s, %s).",
           isDryRun ? "(dry run) " : "", cursorTime, executionTime);
-      tm()
-          .transact(
+      tm().transact(
               () -> {
                 Cursor cursor = ofy().load().key(Cursor.createGlobalKey(RECURRING_BILLING)).now();
                 DateTime currentCursorTime =
@@ -326,7 +326,9 @@ public class ExpandRecurringBillingEventsAction implements Runnable {
                   return;
                 }
                 if (!isDryRun) {
-                  ofy().save().entity(Cursor.createGlobal(RECURRING_BILLING, executionTime));
+                  CursorDao.saveCursor(
+                      Cursor.createGlobal(RECURRING_BILLING, executionTime),
+                      google.registry.schema.cursor.Cursor.GLOBAL);
                 }
               });
     }

core/src/main/java/google/registry/export/sheet/SyncRegistrarsSheet.java

@@ -25,7 +25,6 @@ import static google.registry.model.registrar.RegistrarContact.Type.LEGAL;
 import static google.registry.model.registrar.RegistrarContact.Type.MARKETING;
 import static google.registry.model.registrar.RegistrarContact.Type.TECH;
 import static google.registry.model.registrar.RegistrarContact.Type.WHOIS;
-import static google.registry.model.transaction.TransactionManagerFactory.tm;
 import static google.registry.util.DateTimeUtils.START_OF_TIME;
 
 import com.google.common.base.Joiner;
@@ -37,6 +36,7 @@ import google.registry.model.common.Cursor;
 import google.registry.model.registrar.Registrar;
 import google.registry.model.registrar.RegistrarAddress;
 import google.registry.model.registrar.RegistrarContact;
+import google.registry.schema.cursor.CursorDao;
 import google.registry.util.Clock;
 import google.registry.util.DateTimeUtils;
 import java.io.IOException;
@@ -153,9 +153,9 @@ class SyncRegistrarsSheet {
                   return builder.build();
                 })
             .collect(toImmutableList()));
-    tm()
-        .transact(
-            () -> ofy().save().entity(Cursor.createGlobal(SYNC_REGISTRAR_SHEET, executionTime)));
+    CursorDao.saveCursor(
+        Cursor.createGlobal(SYNC_REGISTRAR_SHEET, executionTime),
+        google.registry.schema.cursor.Cursor.GLOBAL);
   }
 
   private static String convertContacts(

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

@@ -20,6 +20,7 @@ import static google.registry.model.common.EntityGroupRoot.getCrossTldKey;
 import static google.registry.model.ofy.ObjectifyService.ofy;
 import static google.registry.util.DateTimeUtils.START_OF_TIME;
 
+import com.google.common.base.Splitter;
 import com.googlecode.objectify.Key;
 import com.googlecode.objectify.annotation.Entity;
 import com.googlecode.objectify.annotation.Id;
@@ -27,6 +28,7 @@ import com.googlecode.objectify.annotation.Parent;
 import google.registry.model.ImmutableObject;
 import google.registry.model.UpdateAutoTimestamp;
 import google.registry.model.registry.Registry;
+import java.util.List;
 import org.joda.time.DateTime;
 
 /**
@@ -127,6 +129,11 @@ public class Cursor extends ImmutableObject {
     return lastUpdateTime.getTimestamp();
   }
 
+  public CursorType getType() {
+    List<String> id = Splitter.on('_').splitToList(this.id);
+    return CursorType.valueOf(String.join("_", id.subList(1, id.size())));
+  }
+
   /**
    * Checks that the type of the scoped object (or null) matches the required type for the specified
    * cursor (or null, if the cursor is a global cursor).

core/src/main/java/google/registry/rde/EscrowTaskRunner.java

@@ -15,7 +15,6 @@
 package google.registry.rde;
 
 import static google.registry.model.ofy.ObjectifyService.ofy;
-import static google.registry.model.transaction.TransactionManagerFactory.tm;
 
 import com.google.common.flogger.FluentLogger;
 import google.registry.model.common.Cursor;
@@ -24,6 +23,7 @@ import google.registry.model.registry.Registry;
 import google.registry.request.HttpException.NoContentException;
 import google.registry.request.HttpException.ServiceUnavailableException;
 import google.registry.request.lock.LockHandler;
+import google.registry.schema.cursor.CursorDao;
 import google.registry.util.Clock;
 import java.util.concurrent.Callable;
 import javax.inject.Inject;
@@ -99,7 +99,7 @@ class EscrowTaskRunner {
           task.runWithLock(nextRequiredRun);
           DateTime nextRun = nextRequiredRun.plus(interval);
           logger.atInfo().log("Rolling cursor forward to %s.", nextRun);
-          tm().transact(() -> ofy().save().entity(Cursor.create(cursorType, nextRun, registry)));
+          CursorDao.saveCursor(Cursor.create(cursorType, nextRun, registry), registry.getTldStr());
           return null;
         };
     String lockName = String.format("EscrowTaskRunner %s", task.getClass().getSimpleName());

core/src/main/java/google/registry/rde/PendingDepositChecker.java

@@ -27,6 +27,7 @@ import google.registry.model.rde.RdeMode;
 import google.registry.model.registry.Registries;
 import google.registry.model.registry.Registry;
 import google.registry.model.registry.Registry.TldType;
+import google.registry.schema.cursor.CursorDao;
 import google.registry.util.Clock;
 import javax.inject.Inject;
 import org.joda.time.DateTime;
@@ -107,14 +108,14 @@ public final class PendingDepositChecker {
       final Registry registry,
       final CursorType cursorType,
       final DateTime initialValue) {
-    return tm()
-        .transact(
+    return tm().transact(
             () -> {
               Cursor cursor = ofy().load().key(Cursor.createKey(cursorType, registry)).now();
               if (cursor != null) {
                 return cursor.getCursorTime();
               }
-              ofy().save().entity(Cursor.create(cursorType, initialValue, registry));
+              CursorDao.saveCursor(
+                  Cursor.create(cursorType, initialValue, registry), registry.getTldStr());
               return initialValue;
             });
   }

core/src/main/java/google/registry/rde/RdeStagingReducer.java

@@ -40,6 +40,7 @@ import google.registry.model.rde.RdeRevision;
 import google.registry.model.registry.Registry;
 import google.registry.request.RequestParameters;
 import google.registry.request.lock.LockHandler;
+import google.registry.schema.cursor.CursorDao;
 import google.registry.tldconfig.idn.IdnTableEnum;
 import google.registry.util.TaskQueueUtils;
 import google.registry.xjc.rdeheader.XjcRdeHeader;
@@ -198,13 +199,12 @@ public final class RdeStagingReducer extends Reducer<PendingDeposit, DepositFrag
       }
     }
 
-    // Now that we're done, kick off RdeUploadAction and roll forward the cursor transactionally.
+    // Now that we're done, kick off RdeUploadAction and roll the cursor forward.
     if (key.manual()) {
       logger.atInfo().log("Manual operation; not advancing cursor or enqueuing upload task");
       return;
     }
-    tm()
-        .transact(
+    tm().transact(
             () -> {
               Registry registry = Registry.get(tld);
               DateTime position =
@@ -221,7 +221,8 @@ public final class RdeStagingReducer extends Reducer<PendingDeposit, DepositFrag
                   "Partial ordering of RDE deposits broken: %s %s",
                   position,
                   key);
-              ofy().save().entity(Cursor.create(key.cursor(), newPosition, registry)).now();
+              CursorDao.saveCursor(
+                  Cursor.create(key.cursor(), newPosition, registry), registry.getTldStr());
               logger.atInfo().log(
                   "Rolled forward %s on %s cursor to %s", key.cursor(), tld, newPosition);
               RdeRevision.saveRevision(tld, watermark, mode, revision);

core/src/main/java/google/registry/rde/RdeUploadAction.java

@@ -52,6 +52,7 @@ import google.registry.request.Parameter;
 import google.registry.request.RequestParameters;
 import google.registry.request.Response;
 import google.registry.request.auth.Auth;
+import google.registry.schema.cursor.CursorDao;
 import google.registry.util.Clock;
 import google.registry.util.Retrier;
 import google.registry.util.TaskQueueUtils;
@@ -170,12 +171,11 @@ public final class RdeUploadAction implements Runnable, EscrowTask {
         () -> upload(xmlFilename, xmlLength, watermark, name), JSchException.class);
     logger.atInfo().log(
         "Updating RDE cursor '%s' for TLD '%s' following successful upload.", RDE_UPLOAD_SFTP, tld);
-    tm()
-        .transact(
+    tm().transact(
             () -> {
-              Cursor updatedSftpCursor =
-                  Cursor.create(RDE_UPLOAD_SFTP, tm().getTransactionTime(), Registry.get(tld));
-              ofy().save().entity(updatedSftpCursor);
+              CursorDao.saveCursor(
+                  Cursor.create(RDE_UPLOAD_SFTP, tm().getTransactionTime(), Registry.get(tld)),
+                  tld);
             });
     response.setContentType(PLAIN_TEXT_UTF_8);
     response.setPayload(String.format("OK %s %s\n", tld, watermark));

core/src/main/java/google/registry/reporting/icann/IcannReportingUploadAction.java

@@ -44,6 +44,7 @@ import google.registry.request.Parameter;
 import google.registry.request.Response;
 import google.registry.request.auth.Auth;
 import google.registry.request.lock.LockHandler;
+import google.registry.schema.cursor.CursorDao;
 import google.registry.util.Clock;
 import google.registry.util.EmailMessage;
 import google.registry.util.Retrier;
@@ -51,6 +52,7 @@ import google.registry.util.SendEmailService;
 import java.io.IOException;
 import java.io.InputStream;
 import java.util.Map;
+import java.util.Optional;
 import java.util.concurrent.Callable;
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
@@ -186,7 +188,9 @@ public final class IcannReportingUploadAction implements Runnable {
               cursorType,
               cursorTime.withTimeAtStartOfDay().withDayOfMonth(1).plusMonths(1),
               Registry.get(tldStr));
-      tm().transact(() -> ofy().save().entity(newCursor));
+      CursorDao.saveCursor(
+          newCursor,
+          Optional.ofNullable(tldStr).orElse(google.registry.schema.cursor.Cursor.GLOBAL));
     }
   }
 

core/src/main/java/google/registry/schema/cursor/CursorDao.java

@@ -15,8 +15,13 @@
 package google.registry.schema.cursor;
 
 import static com.google.appengine.api.search.checkers.Preconditions.checkNotNull;
+import static google.registry.model.ofy.ObjectifyService.ofy;
 import static google.registry.model.transaction.TransactionManagerFactory.jpaTm;
+import static google.registry.model.transaction.TransactionManagerFactory.tm;
 
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.flogger.FluentLogger;
 import google.registry.model.common.Cursor.CursorType;
 import google.registry.schema.cursor.Cursor.CursorId;
 import java.util.List;
@@ -24,6 +29,8 @@ import java.util.List;
 /** Data access object class for {@link Cursor}. */
 public class CursorDao {
 
+  private static final FluentLogger logger = FluentLogger.forEnclosingClass();
+
   public static void save(Cursor cursor) {
     jpaTm()
         .transact(
@@ -32,9 +39,19 @@ public class CursorDao {
             });
   }
 
+  public static void saveAll(ImmutableSet<Cursor> cursors) {
+    jpaTm()
+        .transact(
+            () -> {
+              for (Cursor cursor : cursors) {
+                jpaTm().getEntityManager().merge(cursor);
+              }
+            });
+  }
+
   public static Cursor load(CursorType type, String scope) {
     checkNotNull(scope, "The scope of the cursor to load cannot be null");
-    checkNotNull(type, "The type of the cursor to load must be specified");
+    checkNotNull(type, "The type of the cursor to load cannot be null");
     return jpaTm()
         .transact(() -> jpaTm().getEntityManager().find(Cursor.class, new CursorId(type, scope)));
   }
@@ -67,4 +84,50 @@ public class CursorDao {
                     .setParameter("type", type)
                     .getResultList());
   }
+
+  /**
+   * This writes the given cursor to Datastore. If the save to Datastore succeeds, then a new
+   * Schema/Cursor object is created and attempted to save to Cloud SQL. If the save to Cloud SQL
+   * fails, the exception is logged, but does not cause the method to fail.
+   */
+  public static void saveCursor(google.registry.model.common.Cursor cursor, String scope) {
+    tm().transact(() -> ofy().save().entity(cursor));
+    CursorType type = cursor.getType();
+    try {
+      Cursor cloudSqlCursor = Cursor.create(type, scope, cursor.getCursorTime());
+      save(cloudSqlCursor);
+    } catch (Exception e) {
+      logger.atSevere().withCause(e).log("Error saving cursor to Cloud SQL.");
+    }
+  }
+
+  /**
+   * This takes in multiple cursors and saves them to Datastore. If those saves succeed, it attempts
+   * to save the cursors to Cloud SQL. If the save to Cloud SQL fails, the exception is logged, but
+   * does not cause the method to fail.
+   */
+  public static void saveCursors(
+      ImmutableMap<google.registry.model.common.Cursor, String> cursors) {
+    // Save the cursors to Datastore
+    tm().transact(
+            () -> {
+              for (google.registry.model.common.Cursor cursor : cursors.keySet()) {
+                ofy().save().entity(cursor);
+              }
+            });
+    // Try to save the cursors to Cloud SQL
+    try {
+      ImmutableSet.Builder<Cursor> cloudSqlCursors = new ImmutableSet.Builder<>();
+      cursors
+          .keySet()
+          .forEach(
+              cursor ->
+                  cloudSqlCursors.add(
+                      Cursor.create(
+                          cursor.getType(), cursors.get(cursor), cursor.getCursorTime())));
+      saveAll(cloudSqlCursors.build());
+    } catch (Exception e) {
+      logger.atSevere().withCause(e).log("Error saving cursor to Cloud SQL.");
+    }
+  }
 }

core/src/main/java/google/registry/tools/UpdateCursorsCommand.java

@@ -14,21 +14,22 @@
 
 package google.registry.tools;
 
-import static google.registry.model.ofy.ObjectifyService.ofy;
 import static google.registry.util.CollectionUtils.isNullOrEmpty;
 
 import com.beust.jcommander.Parameter;
 import com.beust.jcommander.Parameters;
+import com.google.common.collect.ImmutableMap;
 import google.registry.model.common.Cursor;
 import google.registry.model.common.Cursor.CursorType;
 import google.registry.model.registry.Registry;
+import google.registry.schema.cursor.CursorDao;
 import google.registry.tools.params.DateTimeParameter;
 import java.util.List;
 import org.joda.time.DateTime;
 
 /** Modifies {@link Cursor} timestamps used by locking rolling cursor tasks, like in RDE. */
 @Parameters(separators = " =", commandDescription = "Modifies cursor timestamps used by LRC tasks")
-final class UpdateCursorsCommand extends MutatingCommand {
+final class UpdateCursorsCommand extends ConfirmingCommand implements CommandWithCloudSql {
 
   @Parameter(description = "TLDs on which to operate. Omit for global cursors.")
   private List<String> tlds;
@@ -46,19 +47,46 @@ final class UpdateCursorsCommand extends MutatingCommand {
       required = true)
   private DateTime newTimestamp;
 
+  ImmutableMap<Cursor, String> cursorsToUpdate;
+
   @Override
   protected void init() {
+    ImmutableMap.Builder<Cursor, String> cursorsToUpdateBuilder = new ImmutableMap.Builder<>();
     if (isNullOrEmpty(tlds)) {
-      Cursor cursor = ofy().load().key(Cursor.createGlobalKey(cursorType)).now();
-      stageEntityChange(cursor, Cursor.createGlobal(cursorType, newTimestamp));
+      cursorsToUpdateBuilder.put(
+          Cursor.createGlobal(cursorType, newTimestamp),
+          google.registry.schema.cursor.Cursor.GLOBAL);
     } else {
       for (String tld : tlds) {
         Registry registry = Registry.get(tld);
-        Cursor cursor = ofy().load().key(Cursor.createKey(cursorType, registry)).now();
-        stageEntityChange(
-            cursor,
-            Cursor.create(cursorType, newTimestamp, registry));
+        cursorsToUpdateBuilder.put(
+            Cursor.create(cursorType, newTimestamp, registry), registry.getTldStr());
       }
     }
+    cursorsToUpdate = cursorsToUpdateBuilder.build();
+  }
+
+  @Override
+  protected String execute() throws Exception {
+    CursorDao.saveCursors(cursorsToUpdate);
+    return String.format("Updated %d cursors.\n", cursorsToUpdate.size());
+  }
+
+  /** Returns the changes that have been staged thus far. */
+  @Override
+  protected String prompt() {
+    StringBuilder changes = new StringBuilder();
+    if (cursorsToUpdate.isEmpty()) {
+      return "No cursor changes to apply.";
+    }
+    cursorsToUpdate.entrySet().stream()
+        .forEach(entry -> changes.append(getChangeString(entry.getKey(), entry.getValue())));
+    return changes.toString();
+  }
+
+  private String getChangeString(Cursor cursor, String scope) {
+    return String.format(
+        "Change cursorTime of %s for Scope:%s to %s\n",
+        cursor.getType(), scope, cursor.getCursorTime());
   }
 }