Don't retry RDE upload tasks that have failed dependencies

New upload tasks are created every 4 hours, so if we're waiting on a 2 hour SFTP cooldown or some other long-running dependency like generating the RDE report, just delete this task and let it re-run at the next 4 hour period.  No need to let these tasks continue gumming up the queue.

Note that this method of throwing NoContentException to abort the task without enqueuing it for retry is already being used by RdeReportAction for the same purpose.

-------------
Created by MOE: https://github.com/google/moe
MOE_MIGRATED_REVID=201372808

java/google/registry/rde/EscrowTaskRunner.java

@@ -94,15 +94,11 @@ class EscrowTaskRunner {
           if (nextRequiredRun.isAfter(startOfToday)) {
             throw new NoContentException("Already completed");
           }
-          logger.atInfo().log("Cursor: %s", nextRequiredRun);
+          logger.atInfo().log("Current cursor is: %s", nextRequiredRun);
           task.runWithLock(nextRequiredRun);
-          ofy()
+          DateTime nextRun = nextRequiredRun.plus(interval);
-              .transact(
+          logger.atInfo().log("Rolling cursor forward to %s.", nextRun);
-                  () ->
+          ofy().transact(() -> ofy().save().entity(Cursor.create(cursorType, nextRun, registry)));
-                      ofy()
-                          .save()
-                          .entity(
-                              Cursor.create(cursorType, nextRequiredRun.plus(interval), registry)));
           return null;
         };
     String lockName = String.format("EscrowTaskRunner %s", task.getClass().getSimpleName());

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.util.DateTimeUtils.isBeforeOrAt;
 
 import com.google.appengine.tools.cloudstorage.GcsFilename;
 import com.google.common.flogger.FluentLogger;
@@ -82,9 +83,12 @@ public final class RdeReportAction implements Runnable, EscrowTask {
     DateTime cursorTime =
         getCursorTimeOrStartOfTime(
             ofy().load().key(Cursor.createKey(CursorType.RDE_UPLOAD, Registry.get(tld))).now());
-    if (!cursorTime.isAfter(watermark)) {
+    if (isBeforeOrAt(cursorTime, watermark)) {
-      logger.atInfo().log("tld=%s reportCursor=%s uploadCursor=%s", tld, watermark, cursorTime);
+      throw new NoContentException(
-      throw new NoContentException("Waiting for RdeUploadAction to complete");
+          String.format(
+              "Waiting on RdeUploadAction for TLD %s to send %s report; "
+                  + "last upload completion was at %s",
+              tld, watermark, cursorTime));
     }
     String prefix = RdeNamingUtils.makeRydeFilename(tld, watermark, FULL, 1, 0);
     GcsFilename reportFilename = new GcsFilename(bucket, prefix + "-report.xml.ghostryde");

java/google/registry/rde/RdeUploadAction.java

@@ -23,6 +23,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.util.DateTimeUtils.isBeforeOrAt;
 import static java.nio.charset.StandardCharsets.UTF_8;
 import static java.util.Arrays.asList;
 
@@ -46,7 +47,7 @@ import google.registry.model.registry.Registry;
 import google.registry.rde.EscrowTaskRunner.EscrowTask;
 import google.registry.rde.JSchSshSession.JSchSshSessionFactory;
 import google.registry.request.Action;
-import google.registry.request.HttpException.ServiceUnavailableException;
+import google.registry.request.HttpException.NoContentException;
 import google.registry.request.Parameter;
 import google.registry.request.RequestParameters;
 import google.registry.request.Response;
@@ -137,17 +138,27 @@ public final class RdeUploadAction implements Runnable, EscrowTask {
     logger.atInfo().log("Verifying readiness to upload the RDE deposit.");
     DateTime stagingCursorTime = getCursorTimeOrStartOfTime(
         ofy().load().key(Cursor.createKey(CursorType.RDE_STAGING, Registry.get(tld))).now());
-    if (!stagingCursorTime.isAfter(watermark)) {
+    if (isBeforeOrAt(stagingCursorTime, watermark)) {
-      logger.atInfo().log(
+      throw new NoContentException(
-          "tld=%s uploadCursor=%s stagingCursor=%s", tld, watermark, stagingCursorTime);
+          String.format(
-      throw new ServiceUnavailableException("Waiting for RdeStagingAction to complete");
+              "Waiting on RdeStagingAction for TLD %s to send %s upload; "
+                  + "last RDE staging completion was at %s",
+              tld, watermark, stagingCursorTime));
     }
-    DateTime sftpCursorTime = getCursorTimeOrStartOfTime(
+    DateTime sftpCursorTime =
-        ofy().load().key(Cursor.createKey(RDE_UPLOAD_SFTP, Registry.get(tld))).now());
+        getCursorTimeOrStartOfTime(
-    if (sftpCursorTime.plus(sftpCooldown).isAfter(clock.nowUtc())) {
+            ofy().load().key(Cursor.createKey(RDE_UPLOAD_SFTP, Registry.get(tld))).now());
-      // Fail the task good and hard so it retries until the cooldown passes.
+    Duration timeSinceLastSftp = new Duration(sftpCursorTime, clock.nowUtc());
-      logger.atInfo().log("tld=%s cursor=%s sftpCursor=%s", tld, watermark, sftpCursorTime);
+    if (timeSinceLastSftp.isShorterThan(sftpCooldown)) {
-      throw new ServiceUnavailableException("SFTP cooldown has not yet passed");
+      throw new NoContentException(
+          String.format(
+              "Waiting on %d minute SFTP cooldown for TLD %s to send %s upload; "
+                  + "last upload attempt was at %s (%d minutes ago)",
+              sftpCooldown.getStandardMinutes(),
+              tld,
+              watermark,
+              sftpCursorTime,
+              timeSinceLastSftp.getStandardMinutes()));
     }
     int revision = RdeRevision.getNextRevision(tld, watermark, FULL) - 1;
     verify(revision >= 0, "RdeRevision was not set on generated deposit");
@@ -161,22 +172,16 @@ public final class RdeUploadAction implements Runnable, EscrowTask {
     logger.atInfo().log("Commencing RDE upload for TLD '%s' to '%s'.", tld, uploadUrl);
     final long xmlLength = readXmlLength(xmlLengthFilename);
     retrier.callWithRetry(
-        () -> {
+        () -> upload(xmlFilename, xmlLength, watermark, name), JSchException.class);
-          upload(xmlFilename, xmlLength, watermark, name);
-          return null;
-        },
-        JSchException.class);
     logger.atInfo().log(
         "Updating RDE cursor '%s' for TLD '%s' following successful upload.", RDE_UPLOAD_SFTP, tld);
     ofy()
         .transact(
-            () ->
+            () -> {
-                ofy()
+              Cursor updatedSftpCursor =
-                    .save()
+                  Cursor.create(RDE_UPLOAD_SFTP, ofy().getTransactionTime(), Registry.get(tld));
-                    .entity(
+              ofy().save().entity(updatedSftpCursor);
-                        Cursor.create(
+            });
-                            RDE_UPLOAD_SFTP, ofy().getTransactionTime(), Registry.get(tld)))
-                    .now());
     response.setContentType(PLAIN_TEXT_UTF_8);
     response.setPayload(String.format("OK %s %s\n", tld, watermark));
   }

javatests/google/registry/rde/RdeReportActionTest.java

@@ -17,6 +17,8 @@ package google.registry.rde;
 import static com.google.appengine.api.urlfetch.HTTPMethod.PUT;
 import static com.google.common.net.MediaType.PLAIN_TEXT_UTF_8;
 import static com.google.common.truth.Truth.assertThat;
+import static google.registry.model.common.Cursor.CursorType.RDE_REPORT;
+import static google.registry.model.common.Cursor.CursorType.RDE_UPLOAD;
 import static google.registry.model.ofy.ObjectifyService.ofy;
 import static google.registry.testing.DatastoreHelper.createTld;
 import static google.registry.testing.DatastoreHelper.persistResource;
@@ -45,9 +47,9 @@ import com.google.common.collect.ImmutableMap;
 import com.google.common.io.ByteSource;
 import google.registry.gcs.GcsUtils;
 import google.registry.model.common.Cursor;
-import google.registry.model.common.Cursor.CursorType;
 import google.registry.model.registry.Registry;
 import google.registry.request.HttpException.InternalServerErrorException;
+import google.registry.request.HttpException.NoContentException;
 import google.registry.testing.AppEngineRule;
 import google.registry.testing.BouncyCastleProviderRule;
 import google.registry.testing.FakeClock;
@@ -120,9 +122,9 @@ public class RdeReportActionTest {
     PGPPublicKey encryptKey = new FakeKeyringModule().get().getRdeStagingEncryptionKey();
     createTld("test");
     persistResource(
-        Cursor.create(CursorType.RDE_REPORT, DateTime.parse("2006-06-06TZ"), Registry.get("test")));
+        Cursor.create(RDE_REPORT, DateTime.parse("2006-06-06TZ"), Registry.get("test")));
     persistResource(
-        Cursor.create(CursorType.RDE_UPLOAD, DateTime.parse("2006-06-07TZ"), Registry.get("test")));
+        Cursor.create(RDE_UPLOAD, DateTime.parse("2006-06-07TZ"), Registry.get("test")));
     writeGcsFile(
         gcsService,
         reportFile,
@@ -136,7 +138,7 @@ public class RdeReportActionTest {
     action.tld = "lol";
     action.run();
     verify(runner).lockRunAndRollForward(
-        action, Registry.get("lol"), standardSeconds(30), CursorType.RDE_REPORT, standardDays(1));
+        action, Registry.get("lol"), standardSeconds(30), RDE_REPORT, standardDays(1));
     verifyNoMoreInteractions(runner);
   }
 
@@ -166,6 +168,20 @@ public class RdeReportActionTest {
     assertThat(report.getWatermark()).isEqualTo(DateTime.parse("2010-10-17T00:00:00Z"));
   }
 
+  @Test
+  public void testRunWithLock_uploadNotFinished_throws204() {
+    persistResource(
+        Cursor.create(RDE_UPLOAD, DateTime.parse("2006-06-06TZ"), Registry.get("test")));
+    NoContentException thrown =
+        assertThrows(
+            NoContentException.class, () -> createAction().runWithLock(loadRdeReportCursor()));
+    assertThat(thrown)
+        .hasMessageThat()
+        .isEqualTo(
+            "Waiting on RdeUploadAction for TLD test to send 2006-06-06T00:00:00.000Z report; "
+                + "last upload completion was at 2006-06-06T00:00:00.000Z");
+  }
+
   @Test
   public void testRunWithLock_badRequest_throws500WithErrorInfo() throws Exception {
     when(httpResponse.getResponseCode()).thenReturn(SC_BAD_REQUEST);
@@ -202,7 +218,7 @@ public class RdeReportActionTest {
   private DateTime loadRdeReportCursor() {
     return ofy()
         .load()
-        .key(Cursor.createKey(CursorType.RDE_REPORT, Registry.get("test")))
+        .key(Cursor.createKey(RDE_REPORT, Registry.get("test")))
         .now()
         .getCursorTime();
   }

javatests/google/registry/rde/RdeUploadActionTest.java

@@ -17,6 +17,8 @@ package google.registry.rde;
 import static com.google.common.net.MediaType.PLAIN_TEXT_UTF_8;
 import static com.google.common.truth.Truth.assertThat;
 import static com.google.common.truth.Truth.assertWithMessage;
+import static google.registry.model.common.Cursor.CursorType.RDE_STAGING;
+import static google.registry.model.common.Cursor.CursorType.RDE_UPLOAD_SFTP;
 import static google.registry.model.ofy.ObjectifyService.ofy;
 import static google.registry.model.rde.RdeMode.FULL;
 import static google.registry.testing.DatastoreHelper.createTld;
@@ -30,6 +32,7 @@ import static google.registry.testing.TaskQueueHelper.assertNoTasksEnqueued;
 import static google.registry.testing.TaskQueueHelper.assertTasksEnqueued;
 import static java.nio.charset.StandardCharsets.UTF_8;
 import static org.joda.time.Duration.standardDays;
+import static org.joda.time.Duration.standardHours;
 import static org.joda.time.Duration.standardSeconds;
 import static org.junit.Assume.assumeTrue;
 import static org.mockito.Matchers.anyInt;
@@ -58,7 +61,7 @@ import google.registry.model.common.Cursor.CursorType;
 import google.registry.model.rde.RdeRevision;
 import google.registry.model.registry.Registry;
 import google.registry.rde.JSchSshSession.JSchSshSessionFactory;
-import google.registry.request.HttpException.ServiceUnavailableException;
+import google.registry.request.HttpException.NoContentException;
 import google.registry.request.RequestParameters;
 import google.registry.testing.AppEngineRule;
 import google.registry.testing.BouncyCastleProviderRule;
@@ -113,6 +116,7 @@ public class RdeUploadActionTest {
       new GcsFilename("bucket", "tld_2010-10-17_full_S1_R1.xml.length");
   private static final GcsFilename REPORT_R1_FILE =
       new GcsFilename("bucket", "tld_2010-10-17_full_S1_R1-report.xml.ghostryde");
+
   @Rule
   public final SftpServerRule sftpd = new SftpServerRule();
 
@@ -283,8 +287,7 @@ public class RdeUploadActionTest {
     URI uploadUrl = URI.create(String.format("sftp://user:password@localhost:%d/", port));
     DateTime stagingCursor = DateTime.parse("2010-10-18TZ");
     DateTime uploadCursor = DateTime.parse("2010-10-17TZ");
-    persistResource(
+    persistResource(Cursor.create(RDE_STAGING, stagingCursor, Registry.get("tld")));
-        Cursor.create(CursorType.RDE_STAGING, stagingCursor, Registry.get("tld")));
     RdeUploadAction action = createAction(uploadUrl);
     action.lazyJsch = Lazies.of(createThrowingJSchSpy(action.lazyJsch.get(), 2));
     action.runWithLock(uploadCursor);
@@ -304,8 +307,7 @@ public class RdeUploadActionTest {
     URI uploadUrl = URI.create(String.format("sftp://user:password@localhost:%d/", port));
     DateTime stagingCursor = DateTime.parse("2010-10-18TZ");
     DateTime uploadCursor = DateTime.parse("2010-10-17TZ");
-    persistResource(
+    persistResource(Cursor.create(RDE_STAGING, stagingCursor, Registry.get("tld")));
-        Cursor.create(CursorType.RDE_STAGING, stagingCursor, Registry.get("tld")));
     RdeUploadAction action = createAction(uploadUrl);
     action.lazyJsch = Lazies.of(createThrowingJSchSpy(action.lazyJsch.get(), 3));
     RuntimeException thrown =
@@ -319,8 +321,7 @@ public class RdeUploadActionTest {
     URI uploadUrl = URI.create(String.format("sftp://user:password@localhost:%d/", port));
     DateTime stagingCursor = DateTime.parse("2010-10-18TZ");
     DateTime uploadCursor = DateTime.parse("2010-10-17TZ");
-    persistResource(
+    persistResource(Cursor.create(RDE_STAGING, stagingCursor, Registry.get("tld")));
-        Cursor.create(CursorType.RDE_STAGING, stagingCursor, Registry.get("tld")));
     createAction(uploadUrl).runWithLock(uploadCursor);
     assertThat(response.getStatus()).isEqualTo(200);
     assertThat(response.getContentType()).isEqualTo(PLAIN_TEXT_UTF_8);
@@ -343,8 +344,7 @@ public class RdeUploadActionTest {
     URI uploadUrl = URI.create(String.format("sftp://user:password@localhost:%d/", port));
     DateTime stagingCursor = DateTime.parse("2010-10-18TZ");
     DateTime uploadCursor = DateTime.parse("2010-10-17TZ");
-    persistSimpleResource(
+    persistSimpleResource(Cursor.create(RDE_STAGING, stagingCursor, Registry.get("tld")));
-        Cursor.create(CursorType.RDE_STAGING, stagingCursor, Registry.get("tld")));
     createAction(uploadUrl).runWithLock(uploadCursor);
     assertThat(response.getStatus()).isEqualTo(200);
     assertThat(response.getContentType()).isEqualTo(PLAIN_TEXT_UTF_8);
@@ -363,8 +363,7 @@ public class RdeUploadActionTest {
     URI uploadUrl = URI.create(String.format("sftp://user:password@localhost:%d/", port));
     DateTime stagingCursor = DateTime.parse("2010-10-18TZ");
     DateTime uploadCursor = DateTime.parse("2010-10-17TZ");
-    persistResource(
+    persistResource(Cursor.create(RDE_STAGING, stagingCursor, Registry.get("tld")));
-        Cursor.create(CursorType.RDE_STAGING, stagingCursor, Registry.get("tld")));
     createAction(uploadUrl).runWithLock(uploadCursor);
     // Only verify signature for SFTP versions, since we check elsewhere that the GCS files are
     // identical to the ones sent over SFTP.
@@ -378,15 +377,36 @@ public class RdeUploadActionTest {
   }
 
   @Test
-  public void testRunWithLock_stagingNotFinished_throws503() {
+  public void testRunWithLock_stagingNotFinished_throws204() {
+    URI url = URI.create("sftp://user:password@localhost:32323/");
     DateTime stagingCursor = DateTime.parse("2010-10-17TZ");
     DateTime uploadCursor = DateTime.parse("2010-10-17TZ");
-    persistResource(
+    persistResource(Cursor.create(RDE_STAGING, stagingCursor, Registry.get("tld")));
-        Cursor.create(CursorType.RDE_STAGING, stagingCursor, Registry.get("tld")));
+    NoContentException thrown =
-    ServiceUnavailableException thrown =
+        assertThrows(NoContentException.class, () -> createAction(url).runWithLock(uploadCursor));
-        assertThrows(
+    assertThat(thrown)
-            ServiceUnavailableException.class, () -> createAction(null).runWithLock(uploadCursor));
+        .hasMessageThat()
-    assertThat(thrown).hasMessageThat().contains("Waiting for RdeStagingAction to complete");
+        .isEqualTo(
+            "Waiting on RdeStagingAction for TLD tld to send 2010-10-17T00:00:00.000Z upload; "
+                + "last RDE staging completion was at 2010-10-17T00:00:00.000Z");
+  }
+
+  @Test
+  public void testRunWithLock_sftpCooldownNotPassed_throws204() {
+    RdeUploadAction action = createAction(URI.create("sftp://user:password@localhost:32323/"));
+    action.sftpCooldown = standardHours(2);
+    DateTime stagingCursor = DateTime.parse("2010-10-18TZ");
+    DateTime uploadCursor = DateTime.parse("2010-10-17TZ");
+    DateTime sftpCursor = uploadCursor.minusMinutes(97); // Within the 2 hour cooldown period.
+    persistResource(Cursor.create(RDE_STAGING, stagingCursor, Registry.get("tld")));
+    persistResource(Cursor.create(RDE_UPLOAD_SFTP, sftpCursor, Registry.get("tld")));
+    NoContentException thrown =
+        assertThrows(NoContentException.class, () -> action.runWithLock(uploadCursor));
+    assertThat(thrown)
+        .hasMessageThat()
+        .isEqualTo(
+            "Waiting on 120 minute SFTP cooldown for TLD tld to send 2010-10-17T00:00:00.000Z "
+                + "upload; last upload attempt was at 2010-10-16T22:23:00.000Z (97 minutes ago)");
   }
 
   private String slurp(InputStream is) throws FileNotFoundException, IOException {