Add dual read for cursors (#473)

* Add dual read for cursors

* Fix loadAndCompareAll to batch load cursors

* fix javadocs

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

@@ -25,6 +25,8 @@ import static google.registry.model.ofy.ObjectifyService.ofy;
 import static google.registry.model.reporting.HistoryEntry.Type.DOMAIN_AUTORENEW;
 import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
 import static google.registry.pricing.PricingEngineProxy.getDomainRenewCost;
+import static google.registry.schema.cursor.Cursor.GLOBAL;
+import static google.registry.schema.cursor.CursorDao.loadAndCompare;
 import static google.registry.util.CollectionUtils.union;
 import static google.registry.util.DateTimeUtils.START_OF_TIME;
 import static google.registry.util.DateTimeUtils.earliestOf;
@@ -93,6 +95,7 @@ public class ExpandRecurringBillingEventsAction implements Runnable {
   @Override
   public void run() {
     Cursor cursor = ofy().load().key(Cursor.createGlobalKey(RECURRING_BILLING)).now();
+    loadAndCompare(cursor, GLOBAL);
     DateTime executeTime = clock.nowUtc();
     DateTime persistedCursorTime = (cursor == null ? START_OF_TIME : cursor.getCursorTime());
     DateTime cursorTime = cursorTimeParam.orElse(persistedCursorTime);
@@ -317,6 +320,7 @@ public class ExpandRecurringBillingEventsAction implements Runnable {
       tm().transact(
               () -> {
                 Cursor cursor = ofy().load().key(Cursor.createGlobalKey(RECURRING_BILLING)).now();
+                loadAndCompare(cursor, GLOBAL);
                 DateTime currentCursorTime =
                     (cursor == null ? START_OF_TIME : cursor.getCursorTime());
                 if (!currentCursorTime.equals(expectedPersistedCursorTime)) {
@@ -327,8 +331,7 @@ public class ExpandRecurringBillingEventsAction implements Runnable {
                 }
                 if (!isDryRun) {
                   CursorDao.saveCursor(
-                      Cursor.createGlobal(RECURRING_BILLING, executionTime),
+                      Cursor.createGlobal(RECURRING_BILLING, executionTime), GLOBAL);
-                      google.registry.schema.cursor.Cursor.GLOBAL);
                 }
               });
     }

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

@@ -25,6 +25,8 @@ 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.schema.cursor.Cursor.GLOBAL;
+import static google.registry.schema.cursor.CursorDao.loadAndCompare;
 import static google.registry.util.DateTimeUtils.START_OF_TIME;
 
 import com.google.common.base.Joiner;
@@ -62,6 +64,7 @@ class SyncRegistrarsSheet {
    */
   boolean wereRegistrarsModified() {
     Cursor cursor = ofy().load().key(Cursor.createGlobalKey(SYNC_REGISTRAR_SHEET)).now();
+    loadAndCompare(cursor, GLOBAL);
     DateTime lastUpdateTime = (cursor == null) ? START_OF_TIME : cursor.getCursorTime();
     for (Registrar registrar : Registrar.loadAllCached()) {
       if (DateTimeUtils.isAtOrAfter(registrar.getLastUpdateTime(), lastUpdateTime)) {

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

@@ -15,6 +15,7 @@
 package google.registry.rde;
 
 import static google.registry.model.ofy.ObjectifyService.ofy;
+import static google.registry.schema.cursor.CursorDao.loadAndCompare;
 
 import com.google.common.flogger.FluentLogger;
 import google.registry.model.common.Cursor;
@@ -91,6 +92,7 @@ class EscrowTaskRunner {
           logger.atInfo().log("TLD: %s", registry.getTld());
           DateTime startOfToday = clock.nowUtc().withTimeAtStartOfDay();
           Cursor cursor = ofy().load().key(Cursor.createKey(cursorType, registry)).now();
+          loadAndCompare(cursor, registry.getTldStr());
           final DateTime nextRequiredRun = (cursor == null ? startOfToday : cursor.getCursorTime());
           if (nextRequiredRun.isAfter(startOfToday)) {
             throw new NoContentException("Already completed");

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

@@ -17,6 +17,7 @@ package google.registry.rde;
 import static com.google.common.base.Preconditions.checkArgument;
 import static google.registry.model.ofy.ObjectifyService.ofy;
 import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
+import static google.registry.schema.cursor.CursorDao.loadAndCompare;
 import static google.registry.util.DateTimeUtils.isBeforeOrAt;
 
 import com.google.common.collect.ImmutableSetMultimap;
@@ -91,6 +92,7 @@ public final class PendingDepositChecker {
       }
       // Avoid creating a transaction unless absolutely necessary.
       Cursor cursor = ofy().load().key(Cursor.createKey(cursorType, registry)).now();
+      loadAndCompare(cursor, registry.getTldStr());
       DateTime cursorValue = (cursor != null ? cursor.getCursorTime() : startingPoint);
       if (isBeforeOrAt(cursorValue, now)) {
         DateTime watermark = (cursor != null
@@ -111,6 +113,7 @@ public final class PendingDepositChecker {
     return tm().transact(
             () -> {
               Cursor cursor = ofy().load().key(Cursor.createKey(cursorType, registry)).now();
+              loadAndCompare(cursor, registry.getTldStr());
               if (cursor != null) {
                 return cursor.getCursorTime();
               }

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

@@ -20,6 +20,7 @@ import static google.registry.model.common.Cursor.getCursorTimeOrStartOfTime;
 import static google.registry.model.ofy.ObjectifyService.ofy;
 import static google.registry.model.rde.RdeMode.FULL;
 import static google.registry.request.Action.Method.POST;
+import static google.registry.schema.cursor.CursorDao.loadAndCompare;
 import static google.registry.util.DateTimeUtils.isBeforeOrAt;
 
 import com.google.appengine.tools.cloudstorage.GcsFilename;
@@ -76,9 +77,10 @@ public final class RdeReportAction implements Runnable, EscrowTask {
 
   @Override
   public void runWithLock(DateTime watermark) throws Exception {
-    DateTime cursorTime =
+    Cursor cursor =
-        getCursorTimeOrStartOfTime(
+        ofy().load().key(Cursor.createKey(CursorType.RDE_UPLOAD, Registry.get(tld))).now();
-            ofy().load().key(Cursor.createKey(CursorType.RDE_UPLOAD, Registry.get(tld))).now());
+    loadAndCompare(cursor, tld);
+    DateTime cursorTime = getCursorTimeOrStartOfTime(cursor);
     if (isBeforeOrAt(cursorTime, watermark)) {
       throw new NoContentException(
           String.format(

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

@@ -22,6 +22,7 @@ import static com.google.common.base.Verify.verify;
 import static google.registry.model.common.Cursor.getCursorTimeOrStartOfTime;
 import static google.registry.model.ofy.ObjectifyService.ofy;
 import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
+import static google.registry.schema.cursor.CursorDao.loadAndCompare;
 import static java.nio.charset.StandardCharsets.UTF_8;
 
 import com.google.appengine.tools.cloudstorage.GcsFilename;
@@ -207,9 +208,9 @@ public final class RdeStagingReducer extends Reducer<PendingDeposit, DepositFrag
     tm().transact(
             () -> {
               Registry registry = Registry.get(tld);
-              DateTime position =
+              Cursor cursor = ofy().load().key(Cursor.createKey(key.cursor(), registry)).now();
-                  getCursorTimeOrStartOfTime(
+              loadAndCompare(cursor, tld);
-                      ofy().load().key(Cursor.createKey(key.cursor(), registry)).now());
+              DateTime position = getCursorTimeOrStartOfTime(cursor);
               checkState(key.interval() != null, "Interval must be present");
               DateTime newPosition = key.watermark().plus(key.interval());
               if (!position.isBefore(newPosition)) {

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

@@ -24,6 +24,7 @@ import static google.registry.model.ofy.ObjectifyService.ofy;
 import static google.registry.model.rde.RdeMode.FULL;
 import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
 import static google.registry.request.Action.Method.POST;
+import static google.registry.schema.cursor.CursorDao.loadAndCompare;
 import static google.registry.util.DateTimeUtils.isBeforeOrAt;
 import static java.util.Arrays.asList;
 
@@ -132,8 +133,10 @@ public final class RdeUploadAction implements Runnable, EscrowTask {
   @Override
   public void runWithLock(final DateTime watermark) throws Exception {
     logger.atInfo().log("Verifying readiness to upload the RDE deposit.");
-    DateTime stagingCursorTime = getCursorTimeOrStartOfTime(
+    Cursor cursor =
-        ofy().load().key(Cursor.createKey(CursorType.RDE_STAGING, Registry.get(tld))).now());
+        ofy().load().key(Cursor.createKey(CursorType.RDE_STAGING, Registry.get(tld))).now();
+    loadAndCompare(cursor, tld);
+    DateTime stagingCursorTime = getCursorTimeOrStartOfTime(cursor);
     if (isBeforeOrAt(stagingCursorTime, watermark)) {
       throw new NoContentException(
           String.format(
@@ -141,9 +144,10 @@ public final class RdeUploadAction implements Runnable, EscrowTask {
                   + "last RDE staging completion was at %s",
               tld, watermark, stagingCursorTime));
     }
-    DateTime sftpCursorTime =
+    Cursor sftpCursor =
-        getCursorTimeOrStartOfTime(
+        ofy().load().key(Cursor.createKey(RDE_UPLOAD_SFTP, Registry.get(tld))).now();
-            ofy().load().key(Cursor.createKey(RDE_UPLOAD_SFTP, Registry.get(tld))).now());
+    loadAndCompare(sftpCursor, tld);
+    DateTime sftpCursorTime = getCursorTimeOrStartOfTime(sftpCursor);
     Duration timeSinceLastSftp = new Duration(sftpCursorTime, clock.nowUtc());
     if (timeSinceLastSftp.isShorterThan(sftpCooldown)) {
       throw new NoContentException(

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

@@ -21,10 +21,10 @@ import static google.registry.model.common.Cursor.getCursorTimeOrStartOfTime;
 import static google.registry.model.ofy.ObjectifyService.ofy;
 import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
 import static google.registry.request.Action.Method.POST;
+import static google.registry.schema.cursor.CursorDao.loadAndCompareAll;
 import static javax.servlet.http.HttpServletResponse.SC_OK;
 
 import com.google.appengine.tools.cloudstorage.GcsFilename;
-import com.google.auto.value.AutoValue;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
 import com.google.common.flogger.FluentLogger;
@@ -54,7 +54,6 @@ import java.util.Optional;
 import java.util.concurrent.Callable;
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
-import javax.annotation.Nullable;
 import javax.inject.Inject;
 import javax.mail.internet.InternetAddress;
 import org.joda.time.DateTime;
@@ -108,7 +107,7 @@ public final class IcannReportingUploadAction implements Runnable {
         () -> {
           ImmutableMap.Builder<String, Boolean> reportSummaryBuilder = new ImmutableMap.Builder<>();
 
-          ImmutableMap<Cursor, CursorInfo> cursors = loadCursors();
+          ImmutableMap<Cursor, String> cursors = loadCursors();
 
           // If cursor time is before now, upload the corresponding report
           cursors.entrySet().stream()
@@ -118,8 +117,8 @@ public final class IcannReportingUploadAction implements Runnable {
                     DateTime cursorTime = getCursorTimeOrStartOfTime(entry.getKey());
                     uploadReport(
                         cursorTime,
-                        entry.getValue().getType(),
+                        entry.getKey().getType(),
-                        entry.getValue().getTld(),
+                        entry.getValue(),
                         reportSummaryBuilder);
                   });
           // Send email of which reports were uploaded
@@ -205,8 +204,8 @@ public final class IcannReportingUploadAction implements Runnable {
         cursorTimeMinusMonth.monthOfYear().get());
   }
 
-  /** Returns a map of each cursor to the CursorType and tld. */
+  /** Returns a map of each cursor to the tld. */
-  private ImmutableMap<Cursor, CursorInfo> loadCursors() {
+  private ImmutableMap<Cursor, String> loadCursors() {
 
     ImmutableSet<Registry> registries = Registries.getTldEntitiesOfType(TldType.REAL);
 
@@ -220,11 +219,13 @@ public final class IcannReportingUploadAction implements Runnable {
     keys.addAll(transactionKeyMap.keySet());
 
     Map<Key<Cursor>, Cursor> cursorMap = ofy().load().keys(keys.build());
-    ImmutableMap.Builder<Cursor, CursorInfo> cursors = new ImmutableMap.Builder<>();
+    ImmutableMap.Builder<Cursor, String> cursors = new ImmutableMap.Builder<>();
-    defaultNullCursorsToNextMonthAndAddToMap(
+    cursors.putAll(
-        activityKeyMap, CursorType.ICANN_UPLOAD_ACTIVITY, cursorMap, cursors);
+        defaultNullCursorsToNextMonthAndAddToMap(
-    defaultNullCursorsToNextMonthAndAddToMap(
+            activityKeyMap, CursorType.ICANN_UPLOAD_ACTIVITY, cursorMap));
-        transactionKeyMap, CursorType.ICANN_UPLOAD_TX, cursorMap, cursors);
+    cursors.putAll(
+        defaultNullCursorsToNextMonthAndAddToMap(
+            transactionKeyMap, CursorType.ICANN_UPLOAD_TX, cursorMap));
     return cursors.build();
   }
 
@@ -234,15 +235,13 @@ public final class IcannReportingUploadAction implements Runnable {
   }
 
   /**
-   * Populate the cursors map with the Cursor and CursorInfo for each key in the keyMap. If the key
+   * Return a map with the Cursor and scope for each key in the keyMap. If the key from the keyMap
-   * from the keyMap does not have an existing cursor, create a new cursor with a default cursorTime
+   * does not have an existing cursor, create a new cursor with a default cursorTime of the first of
-   * of the first of next month.
+   * next month.
    */
-  private void defaultNullCursorsToNextMonthAndAddToMap(
+  private ImmutableMap<Cursor, String> defaultNullCursorsToNextMonthAndAddToMap(
-      Map<Key<Cursor>, Registry> keyMap,
+      Map<Key<Cursor>, Registry> keyMap, CursorType type, Map<Key<Cursor>, Cursor> cursorMap) {
-      CursorType type,
+    ImmutableMap.Builder<Cursor, String> cursors = new ImmutableMap.Builder<>();
-      Map<Key<Cursor>, Cursor> cursorMap,
-      ImmutableMap.Builder<Cursor, CursorInfo> cursors) {
     keyMap.forEach(
         (key, registry) -> {
           // Cursor time is defaulted to the first of next month since a new tld will not yet have a
@@ -257,8 +256,10 @@ public final class IcannReportingUploadAction implements Runnable {
           if (!cursorMap.containsValue(cursor)) {
             tm().transact(() -> ofy().save().entity(cursor));
           }
-          cursors.put(cursor, CursorInfo.create(type, registry.getTldStr()));
+          cursors.put(cursor, registry.getTldStr());
         });
+    loadAndCompareAll(cursors.build(), type);
+    return cursors.build();
   }
 
   /** Don't retry when reports are already uploaded or can't be uploaded. */
@@ -305,15 +306,4 @@ public final class IcannReportingUploadAction implements Runnable {
         gcsFilename.getBucketName());
   }
 
-  @AutoValue
-  abstract static class CursorInfo {
-    static CursorInfo create(CursorType type, @Nullable String tld) {
-      return new AutoValue_IcannReportingUploadAction_CursorInfo(type, tld);
-    }
-
-    public abstract CursorType getType();
-
-    @Nullable
-    abstract String getTld();
-  }
 }

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

@@ -15,6 +15,7 @@
 package google.registry.schema.cursor;
 
 import static com.google.appengine.api.search.checkers.Preconditions.checkNotNull;
+import static com.google.common.collect.ImmutableMap.toImmutableMap;
 import static google.registry.model.ofy.ObjectifyService.ofy;
 import static google.registry.persistence.transaction.TransactionManagerFactory.jpaTm;
 import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
@@ -25,6 +26,7 @@ import com.google.common.flogger.FluentLogger;
 import google.registry.model.common.Cursor.CursorType;
 import google.registry.schema.cursor.Cursor.CursorId;
 import java.util.List;
+import javax.annotation.Nullable;
 
 /** Data access object class for {@link Cursor}. */
 public class CursorDao {
@@ -130,4 +132,73 @@ public class CursorDao {
       logger.atSevere().withCause(e).log("Error saving cursor to Cloud SQL.");
     }
   }
+
+  /**
+   * Loads in cursor from Cloud SQL and compares it to the Datastore cursor
+   *
+   * <p>This takes in a cursor from Datastore and checks to see if it exists in Cloud SQL and has
+   * the same value. If a difference is detected, or the Cloud SQL cursor does not exist, a warning
+   * is logged.
+   */
+  public static void loadAndCompare(
+      @Nullable google.registry.model.common.Cursor datastoreCursor, String scope) {
+    if (datastoreCursor == null) {
+      return;
+    }
+    try {
+      // Load the corresponding cursor from Cloud SQL
+      Cursor cloudSqlCursor = load(datastoreCursor.getType(), scope);
+      compare(datastoreCursor, cloudSqlCursor, scope);
+    } catch (Throwable t) {
+      logger.atSevere().withCause(t).log("Error comparing cursors.");
+    }
+  }
+
+  /**
+   * Loads in all cursors of a given type from Cloud SQL and compares them to Datastore
+   *
+   * <p>This takes in cursors from Datastore and checks to see if they exists in Cloud SQL and have
+   * the same value. If a difference is detected, or a Cloud SQL cursor does not exist, a warning is
+   * logged.
+   */
+  public static void loadAndCompareAll(
+      ImmutableMap<google.registry.model.common.Cursor, String> cursors, CursorType type) {
+    try {
+      // Load all the cursors of that type from Cloud SQL
+      List<Cursor> cloudSqlCursors = loadByType(type);
+
+      // Create a map of each tld to its cursor if one exists
+      ImmutableMap<String, Cursor> cloudSqlCursorMap =
+          cloudSqlCursors.stream().collect(toImmutableMap(c -> c.getScope(), c -> c));
+
+      // Compare each Datastore cursor with its corresponding Cloud SQL cursor
+      for (google.registry.model.common.Cursor cursor : cursors.keySet()) {
+        Cursor cloudSqlCursor = cloudSqlCursorMap.get(cursors.get(cursor));
+        compare(cursor, cloudSqlCursor, cursors.get(cursor));
+      }
+    } catch (Throwable t) {
+      logger.atSevere().withCause(t).log("Error comparing cursors.");
+    }
+  }
+
+  private static void compare(
+      google.registry.model.common.Cursor datastoreCursor,
+      @Nullable Cursor cloudSqlCursor,
+      String scope) {
+    if (cloudSqlCursor == null) {
+      logger.atWarning().log(
+          String.format(
+              "Cursor of type %s with the scope %s was not found in Cloud SQL.",
+              datastoreCursor.getType().name(), scope));
+    } else if (!datastoreCursor.getCursorTime().equals(cloudSqlCursor.getCursorTime())) {
+      logger.atWarning().log(
+          String.format(
+              "This cursor of type %s with the scope %s has a cursorTime of %s in Datastore and %s"
+                  + " in Cloud SQL.",
+              datastoreCursor.getType(),
+              scope,
+              datastoreCursor.getCursorTime(),
+              cloudSqlCursor.getCursorTime()));
+    }
+  }
 }

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

@@ -16,6 +16,7 @@ package google.registry.tools;
 
 import static com.google.common.collect.ImmutableMap.toImmutableMap;
 import static google.registry.model.ofy.ObjectifyService.ofy;
+import static google.registry.schema.cursor.CursorDao.loadAndCompare;
 
 import com.beust.jcommander.Parameter;
 import com.beust.jcommander.Parameters;
@@ -74,6 +75,9 @@ final class ListCursorsCommand implements CommandWithRemoteApi {
   }
 
   private static String renderLine(String tld, Optional<Cursor> cursor) {
+    if (cursor.isPresent()) {
+      loadAndCompare(cursor.get(), tld);
+    }
     return String.format(
         OUTPUT_FMT,
         tld,

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

@@ -19,6 +19,8 @@ import static google.registry.model.common.Cursor.CursorType.BRDA;
 import static google.registry.model.common.Cursor.CursorType.RDE_UPLOAD;
 import static google.registry.model.common.Cursor.CursorType.RECURRING_BILLING;
 import static google.registry.model.ofy.ObjectifyService.ofy;
+import static google.registry.schema.cursor.Cursor.GLOBAL;
+import static google.registry.schema.cursor.CursorDao.loadAndCompare;
 import static google.registry.testing.DatastoreHelper.createTld;
 import static google.registry.testing.DatastoreHelper.persistActiveDomain;
 import static google.registry.util.DateTimeUtils.START_OF_TIME;
@@ -27,19 +29,30 @@ import static org.junit.Assert.assertThrows;
 import google.registry.model.EntityTestCase;
 import google.registry.model.domain.DomainBase;
 import google.registry.model.registry.Registry;
+import google.registry.persistence.transaction.JpaTestRules;
+import google.registry.persistence.transaction.JpaTestRules.JpaIntegrationWithCoverageRule;
 import google.registry.schema.cursor.CursorDao;
+import google.registry.testing.FakeClock;
 import org.joda.time.DateTime;
+import org.junit.Rule;
 import org.junit.Test;
 
 /** Unit tests for {@link Cursor}. */
 public class CursorTest extends EntityTestCase {
 
+  private final FakeClock fakeClock = new FakeClock(DateTime.parse("2010-10-17TZ"));
+
+  @Rule
+  public final JpaIntegrationWithCoverageRule jpaRule =
+      new JpaTestRules.Builder().withClock(fakeClock).buildIntegrationWithCoverageRule();
+
   @Test
   public void testSuccess_persistScopedCursor() {
     createTld("tld");
     clock.advanceOneMilli();
     final DateTime time = DateTime.parse("2012-07-12T03:30:00.000Z");
-    CursorDao.saveCursor(Cursor.create(RDE_UPLOAD, time, Registry.get("tld")), "tld");
+    Cursor cursor = Cursor.create(RDE_UPLOAD, time, Registry.get("tld"));
+    CursorDao.saveCursor(cursor, "tld");
     assertThat(ofy().load().key(Cursor.createKey(BRDA, Registry.get("tld"))).now()).isNull();
     assertThat(
             ofy()
@@ -48,23 +61,24 @@ public class CursorTest extends EntityTestCase {
                 .now()
                 .getCursorTime())
         .isEqualTo(time);
+    loadAndCompare(cursor, "tld");
   }
 
   @Test
   public void testSuccess_persistGlobalCursor() {
     final DateTime time = DateTime.parse("2012-07-12T03:30:00.000Z");
-    CursorDao.saveCursor(
+    CursorDao.saveCursor(Cursor.createGlobal(RECURRING_BILLING, time), GLOBAL);
-        Cursor.createGlobal(RECURRING_BILLING, time), google.registry.schema.cursor.Cursor.GLOBAL);
     assertThat(ofy().load().key(Cursor.createGlobalKey(RECURRING_BILLING)).now().getCursorTime())
         .isEqualTo(time);
+    loadAndCompare(Cursor.createGlobal(RECURRING_BILLING, time), GLOBAL);
   }
 
   @Test
   public void testIndexing() throws Exception {
     final DateTime time = DateTime.parse("2012-07-12T03:30:00.000Z");
-    CursorDao.saveCursor(
+    CursorDao.saveCursor(Cursor.createGlobal(RECURRING_BILLING, time), GLOBAL);
-        Cursor.createGlobal(RECURRING_BILLING, time), google.registry.schema.cursor.Cursor.GLOBAL);
     Cursor cursor = ofy().load().key(Cursor.createGlobalKey(RECURRING_BILLING)).now();
+    loadAndCompare(cursor, GLOBAL);
     verifyIndexing(cursor);
   }
 

core/src/test/java/google/registry/schema/cursor/CursorDaoTest.java

@@ -259,4 +259,178 @@ public class CursorDaoTest {
     assertThat(createdCursor3.getCursorTime()).isEqualTo(cursor3.getCursorTime());
     assertThat(cursor3).isEqualTo(dataStoreCursor3);
   }
+
+  @Test
+  public void loadAndCompare_worksSuccessfully() {
+    loggerToIntercept.addHandler(logHandler);
+    createTld("tld");
+    google.registry.model.common.Cursor cursor =
+        google.registry.model.common.Cursor.create(
+            CursorType.ICANN_UPLOAD_ACTIVITY, fakeClock.nowUtc(), Registry.get("tld"));
+    CursorDao.saveCursor(cursor, "tld");
+    CursorDao.loadAndCompare(cursor, "tld");
+    assertAboutLogs().that(logHandler).hasNoLogsAtLevel(Level.WARNING);
+  }
+
+  @Test
+  public void loadAndCompare_worksSuccessfullyGlobalCursor() {
+    loggerToIntercept.addHandler(logHandler);
+    google.registry.model.common.Cursor cursor =
+        google.registry.model.common.Cursor.createGlobal(
+            CursorType.RECURRING_BILLING, fakeClock.nowUtc());
+    CursorDao.saveCursor(cursor, Cursor.GLOBAL);
+    CursorDao.loadAndCompare(cursor, Cursor.GLOBAL);
+    assertAboutLogs().that(logHandler).hasNoLogsAtLevel(Level.WARNING);
+  }
+
+  @Test
+  public void loadAndCompare_worksSuccessfullyCursorNotInCloudSql() {
+    loggerToIntercept.addHandler(logHandler);
+    createTld("tld");
+    google.registry.model.common.Cursor cursor =
+        google.registry.model.common.Cursor.create(
+            CursorType.ICANN_UPLOAD_ACTIVITY, fakeClock.nowUtc(), Registry.get("tld"));
+    CursorDao.loadAndCompare(cursor, "tld");
+    assertAboutLogs()
+        .that(logHandler)
+        .hasLogAtLevelWithMessage(
+            Level.WARNING,
+            "Cursor of type ICANN_UPLOAD_ACTIVITY with the scope tld was not found in Cloud SQL.");
+  }
+
+  @Test
+  public void loadAndCompare_worksSuccessfullyGlobalCursorNotInCloudSql() {
+    loggerToIntercept.addHandler(logHandler);
+    google.registry.model.common.Cursor cursor =
+        google.registry.model.common.Cursor.createGlobal(
+            CursorType.RECURRING_BILLING, fakeClock.nowUtc());
+    CursorDao.loadAndCompare(cursor, Cursor.GLOBAL);
+    assertAboutLogs()
+        .that(logHandler)
+        .hasLogAtLevelWithMessage(
+            Level.WARNING,
+            "Cursor of type RECURRING_BILLING with the scope GLOBAL was not found in Cloud SQL.");
+  }
+
+  @Test
+  public void loadAndCompare_worksSuccessfullyCursorsNotEqual() {
+    loggerToIntercept.addHandler(logHandler);
+    createTld("tld");
+    google.registry.model.common.Cursor cursor =
+        google.registry.model.common.Cursor.create(
+            CursorType.ICANN_UPLOAD_ACTIVITY, fakeClock.nowUtc(), Registry.get("tld"));
+    CursorDao.saveCursor(cursor, "tld");
+    cursor =
+        google.registry.model.common.Cursor.create(
+            CursorType.ICANN_UPLOAD_ACTIVITY, fakeClock.nowUtc().minusDays(5), Registry.get("tld"));
+    CursorDao.loadAndCompare(cursor, "tld");
+    assertAboutLogs()
+        .that(logHandler)
+        .hasLogAtLevelWithMessage(
+            Level.WARNING,
+            "This cursor of type ICANN_UPLOAD_ACTIVITY with the scope tld has a cursorTime of"
+                + " 1969-12-27T00:00:00.000Z in Datastore and 1970-01-01T00:00:00.000Z in Cloud"
+                + " SQL.");
+  }
+
+  @Test
+  public void loadAndCompareAll_worksSuccessfully() {
+    loggerToIntercept.addHandler(logHandler);
+
+    // Create Datastore cursors
+    createTlds("tld", "foo");
+    google.registry.model.common.Cursor cursor1 =
+        google.registry.model.common.Cursor.create(
+            CursorType.ICANN_UPLOAD_ACTIVITY, fakeClock.nowUtc(), Registry.get("tld"));
+    google.registry.model.common.Cursor cursor2 =
+        google.registry.model.common.Cursor.create(
+            CursorType.ICANN_UPLOAD_ACTIVITY, fakeClock.nowUtc(), Registry.get("foo"));
+
+    // Save cursors to Cloud SQL
+    ImmutableMap<google.registry.model.common.Cursor, String> cursors =
+        ImmutableMap.<google.registry.model.common.Cursor, String>builder()
+            .put(cursor1, "tld")
+            .put(cursor2, "foo")
+            .build();
+    CursorDao.saveCursors(cursors);
+
+    CursorDao.loadAndCompareAll(cursors, CursorType.ICANN_UPLOAD_ACTIVITY);
+    assertAboutLogs().that(logHandler).hasNoLogsAtLevel(Level.WARNING);
+  }
+
+  @Test
+  public void loadAndCompareAll_worksSuccessfullyMissingOne() {
+    loggerToIntercept.addHandler(logHandler);
+
+    // Create Datastore cursors
+    createTlds("tld", "foo", "lol");
+    google.registry.model.common.Cursor cursor1 =
+        google.registry.model.common.Cursor.create(
+            CursorType.ICANN_UPLOAD_ACTIVITY, fakeClock.nowUtc(), Registry.get("tld"));
+    google.registry.model.common.Cursor cursor2 =
+        google.registry.model.common.Cursor.create(
+            CursorType.ICANN_UPLOAD_ACTIVITY, fakeClock.nowUtc(), Registry.get("foo"));
+
+    // Save Cursors to Cloud SQL
+    ImmutableMap.Builder<google.registry.model.common.Cursor, String> cursors =
+        ImmutableMap.<google.registry.model.common.Cursor, String>builder()
+            .put(cursor1, "tld")
+            .put(cursor2, "foo");
+    CursorDao.saveCursors(cursors.build());
+
+    // Create a new Datastore cursor that is not saved to Cloud SQL
+    google.registry.model.common.Cursor cursor3 =
+        google.registry.model.common.Cursor.create(
+            CursorType.ICANN_UPLOAD_ACTIVITY, fakeClock.nowUtc().minusDays(4), Registry.get("lol"));
+
+    // Call loadAndCompareAll with all three Datastore cursors
+    CursorDao.loadAndCompareAll(
+        cursors.put(cursor3, "lol").build(), CursorType.ICANN_UPLOAD_ACTIVITY);
+
+    assertAboutLogs()
+        .that(logHandler)
+        .hasLogAtLevelWithMessage(
+            Level.WARNING,
+            "Cursor of type ICANN_UPLOAD_ACTIVITY with the scope lol was not found in Cloud SQL.");
+  }
+
+  @Test
+  public void loadAndCompareAll_worksSuccessfullyOneWithWrongTime() {
+    loggerToIntercept.addHandler(logHandler);
+
+    // Create Datastore cursors
+    createTlds("tld", "foo");
+    google.registry.model.common.Cursor cursor1 =
+        google.registry.model.common.Cursor.create(
+            CursorType.ICANN_UPLOAD_ACTIVITY, fakeClock.nowUtc(), Registry.get("tld"));
+    google.registry.model.common.Cursor cursor2 =
+        google.registry.model.common.Cursor.create(
+            CursorType.ICANN_UPLOAD_ACTIVITY, fakeClock.nowUtc(), Registry.get("foo"));
+
+    // Save Cursors to Cloud SQL
+    CursorDao.saveCursors(ImmutableMap.of(cursor1, "tld", cursor2, "foo"));
+
+    // Change time of first Datastore cursor, but don't save new time to Cloud SQL
+    google.registry.model.common.Cursor cursor3 =
+        google.registry.model.common.Cursor.create(
+            CursorType.ICANN_UPLOAD_ACTIVITY, fakeClock.nowUtc().minusDays(4), Registry.get("tld"));
+
+    CursorDao.loadAndCompareAll(
+        ImmutableMap.of(cursor2, "foo", cursor3, "tld"), CursorType.ICANN_UPLOAD_ACTIVITY);
+
+    assertAboutLogs()
+        .that(logHandler)
+        .hasLogAtLevelWithMessage(
+            Level.WARNING,
+            "This cursor of type ICANN_UPLOAD_ACTIVITY with the scope tld has a cursorTime of"
+                + " 1969-12-28T00:00:00.000Z in Datastore and 1970-01-01T00:00:00.000Z in Cloud"
+                + " SQL.");
+  }
+
+  @Test
+  public void loadAndCompareAll_worksSuccessfullyEmptyMap() {
+    loggerToIntercept.addHandler(logHandler);
+    CursorDao.loadAndCompareAll(ImmutableMap.of(), CursorType.ICANN_UPLOAD_ACTIVITY);
+    assertAboutLogs().that(logHandler).hasNoLogsAtLevel(Level.WARNING);
+  }
 }

core/src/test/java/google/registry/schema/integration/SqlIntegrationTestSuite.java

@@ -15,6 +15,7 @@
 package google.registry.schema.integration;
 
 import com.google.common.truth.Expect;
+import google.registry.model.common.CursorTest;
 import google.registry.model.domain.DomainBaseSqlTest;
 import google.registry.model.registry.RegistryLockDaoTest;
 import google.registry.persistence.transaction.JpaEntityCoverage;
@@ -60,6 +61,7 @@ import org.junit.runners.Suite.SuiteClasses;
   CreateRegistrarCommandTest.class,
   CreateReservedListCommandTest.class,
   CursorDaoTest.class,
+  CursorTest.class,
   DomainLockUtilsTest.class,
   LockDaoTest.class,
   LockDomainCommandTest.class,