Add support for delay of duration when scheduling a task (#1493)

* Add support for delay by duration when scheduling task

* Fix comments

* Add test for negative duration

* Change delay parameter type to duration

core/src/test/java/google/registry/testing/CloudTasksHelper.java

@@ -39,6 +39,7 @@ import com.google.common.collect.Multimaps;
 import com.google.common.net.HttpHeaders;
 import com.google.common.net.MediaType;
 import com.google.common.truth.Truth8;
+import com.google.protobuf.Timestamp;
 import google.registry.model.ImmutableObject;
 import google.registry.util.CloudTasksUtils;
 import google.registry.util.Retrier;
@@ -195,6 +196,7 @@ public class CloudTasksHelper implements Serializable {
     // tests.
     HttpMethod method = HttpMethod.POST;
     String url;
+    Timestamp scheduleTime;
     Multimap<String, String> headers = ArrayListMultimap.create();
     Multimap<String, String> params = ArrayListMultimap.create();
 
@@ -216,6 +218,7 @@ public class CloudTasksHelper implements Serializable {
           Ascii.toLowerCase(task.getAppEngineHttpRequest().getAppEngineRouting().getService());
       method = task.getAppEngineHttpRequest().getHttpMethod();
       url = uri.getPath();
+      scheduleTime = task.getScheduleTime();
       ImmutableMultimap.Builder<String, String> headerBuilder = new ImmutableMultimap.Builder<>();
       task.getAppEngineHttpRequest()
           .getHeadersMap()
@@ -251,6 +254,7 @@ public class CloudTasksHelper implements Serializable {
       builder.put("url", url);
       builder.put("headers", headers);
       builder.put("params", params);
+      builder.put("scheduleTime", scheduleTime);
       return Maps.filterValues(builder, not(in(asList(null, "", Collections.EMPTY_MAP))));
     }
   }
@@ -293,6 +297,11 @@ public class CloudTasksHelper implements Serializable {
       return this;
     }
 
+    public TaskMatcher scheduleTime(Timestamp scheduleTime) {
+      expected.scheduleTime = scheduleTime;
+      return this;
+    }
+
     public TaskMatcher param(String key, String value) {
       checkNotNull(value, "Test error: A param can never have a null value, so don't assert it");
       expected.params.put(key, value);
@@ -316,6 +325,8 @@ public class CloudTasksHelper implements Serializable {
      *
      * <p>Match fails if any headers or params expected on the TaskMatcher are not found on the
      * Task. Note that the inverse is not true (i.e. there may be extra headers on the Task).
+     *
+     * <p>Schedule time by default is Timestamp.getDefaultInstance() or null.
      */
     @Override
     public boolean test(@Nonnull Task task) {
@@ -324,6 +335,8 @@ public class CloudTasksHelper implements Serializable {
           && (expected.url == null || Objects.equals(expected.url, actual.url))
           && (expected.method == null || Objects.equals(expected.method, actual.method))
           && (expected.service == null || Objects.equals(expected.service, actual.service))
+          && (expected.scheduleTime == null
+              || Objects.equals(expected.scheduleTime, actual.scheduleTime))
           && containsEntries(actual.params, expected.params)
           && containsEntries(actual.headers, expected.headers);
     }

util/src/main/java/google/registry/util/CloudTasksUtils.java

@@ -43,6 +43,7 @@ import java.util.Arrays;
 import java.util.Optional;
 import java.util.Random;
 import java.util.function.Supplier;
+import org.joda.time.Duration;
 
 /** Utilities for dealing with Cloud Tasks. */
 public class CloudTasksUtils implements Serializable {
@@ -167,12 +168,45 @@ public class CloudTasksUtils implements Serializable {
     if (!jitterSeconds.isPresent() || jitterSeconds.get() <= 0) {
       return createTask(path, method, service, params);
     }
-    Instant scheduleTime =
+    return createTask(
-        Instant.ofEpochMilli(
+        path,
-            clock
+        method,
-                .nowUtc()
+        service,
-                .plusMillis(random.nextInt((int) SECONDS.toMillis(jitterSeconds.get())))
+        params,
-                .getMillis());
+        clock,
+        Duration.millis(random.nextInt((int) SECONDS.toMillis(jitterSeconds.get()))));
+  }
+
+  /**
+   * Create a {@link Task} to be enqueued with delay of {@code duration}.
+   *
+   * @param path the relative URI (staring with a slash and ending without one).
+   * @param method the HTTP method to be used for the request, only GET and POST are supported.
+   * @param service the App Engine service to route the request to. Note that with App Engine Task
+   *     Queue API if no service is specified, the service which enqueues the task will be used to
+   *     process the task. Cloud Tasks API does not support this feature so the service will always
+   *     needs to be explicitly specified.
+   * @param params a multi-map of URL query parameters. Duplicate keys are saved as is, and it is up
+   *     to the server to process the duplicate keys.
+   * @param clock a source of time.
+   * @param delay the amount of time that a task needs to delayed for.
+   * @return the enqueued task.
+   * @see <a
+   *     href=ttps://cloud.google.com/appengine/docs/standard/java/taskqueue/push/creating-tasks#target>Specifyinig
+   *     the worker service</a>
+   */
+  private static Task createTask(
+      String path,
+      HttpMethod method,
+      String service,
+      Multimap<String, String> params,
+      Clock clock,
+      Duration delay) {
+    if (delay.isEqual(Duration.ZERO)) {
+      return createTask(path, method, service, params);
+    }
+    checkArgument(delay.isLongerThan(Duration.ZERO), "Negative duration is not supported.");
+    Instant scheduleTime = Instant.ofEpochMilli(clock.nowUtc().getMillis() + delay.getMillis());
     return Task.newBuilder(createTask(path, method, service, params))
         .setScheduleTime(
             Timestamp.newBuilder()
@@ -214,6 +248,18 @@ public class CloudTasksUtils implements Serializable {
     return createTask(path, HttpMethod.GET, service, params, clock, jitterSeconds);
   }
 
+  /** Create a {@link Task} via HTTP.POST that will be delayed for {@code delay}. */
+  public static Task createPostTask(
+      String path, String service, Multimap<String, String> params, Clock clock, Duration delay) {
+    return createTask(path, HttpMethod.POST, service, params, clock, delay);
+  }
+
+  /** Create a {@link Task} via HTTP.GET that will be delayed for {@code delay}. */
+  public static Task createGetTask(
+      String path, String service, Multimap<String, String> params, Clock clock, Duration delay) {
+    return createTask(path, HttpMethod.GET, service, params, clock, delay);
+  }
+
   public abstract static class SerializableCloudTasksClient implements Serializable {
     public abstract Task enqueue(String projectId, String locationId, String queueName, Task task);
   }

util/src/test/java/google/registry/util/CloudTasksUtilsTest.java

@@ -34,6 +34,7 @@ import java.nio.charset.StandardCharsets;
 import java.time.Instant;
 import java.util.Optional;
 import org.joda.time.DateTime;
+import org.joda.time.Duration;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
@@ -219,6 +220,87 @@ public class CloudTasksUtilsTest {
     assertThat(task.getScheduleTime().getSeconds()).isEqualTo(0);
   }
 
+  @Test
+  void testSuccess_createGetTasks_withDelay() {
+    Task task =
+        CloudTasksUtils.createGetTask(
+            "/the/path", "myservice", params, clock, Duration.standardMinutes(10));
+    assertThat(task.getAppEngineHttpRequest().getHttpMethod()).isEqualTo(HttpMethod.GET);
+    assertThat(task.getAppEngineHttpRequest().getRelativeUri())
+        .isEqualTo("/the/path?key1=val1&key2=val2&key1=val3");
+    assertThat(task.getAppEngineHttpRequest().getAppEngineRouting().getService())
+        .isEqualTo("myservice");
+    assertThat(Instant.ofEpochSecond(task.getScheduleTime().getSeconds()))
+        .isEqualTo(Instant.ofEpochMilli(clock.nowUtc().plusMinutes(10).getMillis()));
+  }
+
+  @Test
+  void testSuccess_createPostTasks_withDelay() {
+    Task task =
+        CloudTasksUtils.createPostTask(
+            "/the/path", "myservice", params, clock, Duration.standardMinutes(10));
+    assertThat(task.getAppEngineHttpRequest().getHttpMethod()).isEqualTo(HttpMethod.POST);
+    assertThat(task.getAppEngineHttpRequest().getRelativeUri()).isEqualTo("/the/path");
+    assertThat(task.getAppEngineHttpRequest().getAppEngineRouting().getService())
+        .isEqualTo("myservice");
+    assertThat(task.getAppEngineHttpRequest().getHeadersMap().get("Content-Type"))
+        .isEqualTo("application/x-www-form-urlencoded");
+    assertThat(task.getAppEngineHttpRequest().getBody().toString(StandardCharsets.UTF_8))
+        .isEqualTo("key1=val1&key2=val2&key1=val3");
+    assertThat(task.getScheduleTime().getSeconds()).isNotEqualTo(0);
+    assertThat(Instant.ofEpochSecond(task.getScheduleTime().getSeconds()))
+        .isEqualTo(Instant.ofEpochMilli(clock.nowUtc().plusMinutes(10).getMillis()));
+  }
+
+  @Test
+  void testFailure_createGetTasks_withNegativeDelay() {
+    IllegalArgumentException thrown =
+        assertThrows(
+            IllegalArgumentException.class,
+            () ->
+                CloudTasksUtils.createGetTask(
+                    "/the/path", "myservice", params, clock, Duration.standardMinutes(-10)));
+    assertThat(thrown).hasMessageThat().isEqualTo("Negative duration is not supported.");
+  }
+
+  @Test
+  void testFailure_createPostTasks_withNegativeDelay() {
+    IllegalArgumentException thrown =
+        assertThrows(
+            IllegalArgumentException.class,
+            () ->
+                CloudTasksUtils.createGetTask(
+                    "/the/path", "myservice", params, clock, Duration.standardMinutes(-10)));
+    assertThat(thrown).hasMessageThat().isEqualTo("Negative duration is not supported.");
+  }
+
+  @Test
+  void testSuccess_createPostTasks_withZeroDelay() {
+    Task task =
+        CloudTasksUtils.createPostTask("/the/path", "myservice", params, clock, Duration.ZERO);
+    assertThat(task.getAppEngineHttpRequest().getHttpMethod()).isEqualTo(HttpMethod.POST);
+    assertThat(task.getAppEngineHttpRequest().getRelativeUri()).isEqualTo("/the/path");
+    assertThat(task.getAppEngineHttpRequest().getAppEngineRouting().getService())
+        .isEqualTo("myservice");
+    assertThat(task.getAppEngineHttpRequest().getHeadersMap().get("Content-Type"))
+        .isEqualTo("application/x-www-form-urlencoded");
+    assertThat(task.getAppEngineHttpRequest().getBody().toString(StandardCharsets.UTF_8))
+        .isEqualTo("key1=val1&key2=val2&key1=val3");
+    assertThat(task.getScheduleTime().getSeconds()).isEqualTo(0);
+  }
+
+  @Test
+  void testSuccess_createGetTasks_withZeroDelay() {
+    Task task =
+        CloudTasksUtils.createGetTask("/the/path", "myservice", params, clock, Duration.ZERO);
+    assertThat(task.getAppEngineHttpRequest().getHttpMethod()).isEqualTo(HttpMethod.GET);
+    assertThat(task.getAppEngineHttpRequest().getRelativeUri())
+        .isEqualTo("/the/path?key1=val1&key2=val2&key1=val3");
+    assertThat(task.getAppEngineHttpRequest().getAppEngineRouting().getService())
+        .isEqualTo("myservice");
+    assertThat(task.getScheduleTime().getSeconds()).isEqualTo(0);
+  }
+
   @Test
   void testFailure_illegalPath() {
     assertThrows(