feat(cron) implement Deno.cron() (#21019)

This PR adds unstable `Deno.cron` API to trigger execution of cron jobs.

* State: All cron state is in memory. Cron jobs are scheduled according
to the cron schedule expression and the current time. No state is
persisted to disk.
* Time zone: Cron expressions specify time in UTC.
* Overlapping executions: not permitted. If the next scheduled execution
time occurs while the same cron job is still executing, the scheduled
execution is skipped.
* Retries: failed jobs are automatically retried until they succeed or
until retry threshold is reached. Retry policy can be optionally
specified using `options.backoffSchedule`.

Cargo.toml

@@ -13,6 +13,7 @@ members = [
   "ext/broadcast_channel",
   "ext/cache",
   "ext/console",
+  "ext/cron",
   "ext/crypto",
   "ext/fetch",
   "ext/ffi",
@@ -56,6 +57,7 @@ denokv_remote = "0.2.3"
 deno_broadcast_channel = { version = "0.115.0", path = "./ext/broadcast_channel" }
 deno_cache = { version = "0.53.0", path = "./ext/cache" }
 deno_console = { version = "0.121.0", path = "./ext/console" }
+deno_cron = { version = "0.1.0", path = "./ext/cron" }
 deno_crypto = { version = "0.135.0", path = "./ext/crypto" }
 deno_fetch = { version = "0.145.0", path = "./ext/fetch" }
 deno_ffi = { version = "0.108.0", path = "./ext/ffi" }
@@ -133,6 +135,7 @@ rustls-webpki = "0.101.4"
 rustls-native-certs = "0.6.2"
 webpki-roots = "0.25.2"
 scopeguard = "1.2.0"
+saffron = "=0.1.0"
 serde = { version = "1.0.149", features = ["derive"] }
 serde_bytes = "0.11"
 serde_json = "1.0.85"

cli/build.rs

@@ -349,6 +349,7 @@ deno_core::extension!(
 fn create_cli_snapshot(snapshot_path: PathBuf) -> CreateSnapshotOutput {
   use deno_core::Extension;
   use deno_runtime::deno_cache::SqliteBackedCache;
+  use deno_runtime::deno_cron::local::LocalCronHandler;
   use deno_runtime::deno_http::DefaultHttpPropertyExtractor;
   use deno_runtime::deno_kv::sqlite::SqliteDbHandler;
   use deno_runtime::permissions::PermissionsContainer;
@@ -383,6 +384,7 @@ fn create_cli_snapshot(snapshot_path: PathBuf) -> CreateSnapshotOutput {
     deno_kv::deno_kv::init_ops(SqliteDbHandler::<PermissionsContainer>::new(
       None, None,
     )),
+    deno_cron::deno_cron::init_ops(LocalCronHandler::new()),
     deno_napi::deno_napi::init_ops::<PermissionsContainer>(),
     deno_http::deno_http::init_ops::<DefaultHttpPropertyExtractor>(),
     deno_io::deno_io::init_ops(Default::default()),

cli/tests/integration/js_unit_tests.rs

@@ -24,6 +24,7 @@ util::unit_test_factory!(
     console_test,
     copy_file_test,
     custom_event_test,
+    cron_test,
     dir_test,
     dom_exception_test,
     error_stack_test,

cli/tests/unit/cron_test.ts

@@ -0,0 +1,242 @@
+// Copyright 2018-2023 the Deno authors. All rights reserved. MIT license.
+import { assertEquals, assertThrows, deferred } from "./test_util.ts";
+
+const sleep = (time: number) => new Promise((r) => setTimeout(r, time));
+
+Deno.test(function noNameTest() {
+  assertThrows(
+    // @ts-ignore test
+    () => Deno.cron(),
+    TypeError,
+    "Deno.cron requires a unique name",
+  );
+});
+
+Deno.test(function noSchedule() {
+  assertThrows(
+    // @ts-ignore test
+    () => Deno.cron("foo"),
+    TypeError,
+    "Deno.cron requires a valid schedule",
+  );
+});
+
+Deno.test(function noHandler() {
+  assertThrows(
+    // @ts-ignore test
+    () => Deno.cron("foo", "*/1 * * * *"),
+    TypeError,
+    "Deno.cron requires a handler",
+  );
+});
+
+Deno.test(function invalidNameTest() {
+  assertThrows(
+    () => Deno.cron("abc[]", "*/1 * * * *", () => {}),
+    TypeError,
+    "Invalid cron name",
+  );
+  assertThrows(
+    () => Deno.cron("a**bc", "*/1 * * * *", () => {}),
+    TypeError,
+    "Invalid cron name",
+  );
+  assertThrows(
+    () => Deno.cron("abc<>", "*/1 * * * *", () => {}),
+    TypeError,
+    "Invalid cron name",
+  );
+  assertThrows(
+    () => Deno.cron(";']", "*/1 * * * *", () => {}),
+    TypeError,
+    "Invalid cron name",
+  );
+  assertThrows(
+    () =>
+      Deno.cron(
+        "0000000000000000000000000000000000000000000000000000000000000000000000",
+        "*/1 * * * *",
+        () => {},
+      ),
+    TypeError,
+    "Cron name is too long",
+  );
+});
+
+Deno.test(function invalidScheduleTest() {
+  assertThrows(
+    () => Deno.cron("abc", "bogus", () => {}),
+    TypeError,
+    "Invalid cron schedule",
+  );
+  assertThrows(
+    () => Deno.cron("abc", "* * * * * *", () => {}),
+    TypeError,
+    "Invalid cron schedule",
+  );
+  assertThrows(
+    () => Deno.cron("abc", "* * * *", () => {}),
+    TypeError,
+    "Invalid cron schedule",
+  );
+  assertThrows(
+    () => Deno.cron("abc", "m * * * *", () => {}),
+    TypeError,
+    "Invalid cron schedule",
+  );
+});
+
+Deno.test(function invalidBackoffScheduleTest() {
+  assertThrows(
+    () =>
+      Deno.cron("abc", "*/1 * * * *", () => {}, {
+        backoffSchedule: [1, 1, 1, 1, 1, 1],
+      }),
+    TypeError,
+    "Invalid backoff schedule",
+  );
+  assertThrows(
+    () =>
+      Deno.cron("abc", "*/1 * * * *", () => {}, {
+        backoffSchedule: [3600001],
+      }),
+    TypeError,
+    "Invalid backoff schedule",
+  );
+});
+
+Deno.test(async function tooManyCrons() {
+  const crons: Promise<void>[] = [];
+  const ac = new AbortController();
+  for (let i = 0; i <= 100; i++) {
+    const c = Deno.cron(`abc_${i}`, "*/1 * * * *", () => {}, {
+      signal: ac.signal,
+    });
+    crons.push(c);
+  }
+
+  try {
+    assertThrows(
+      () => {
+        Deno.cron("next-cron", "*/1 * * * *", () => {}, { signal: ac.signal });
+      },
+      TypeError,
+      "Too many crons",
+    );
+  } finally {
+    ac.abort();
+    for (const c of crons) {
+      await c;
+    }
+  }
+});
+
+Deno.test(async function duplicateCrons() {
+  const ac = new AbortController();
+  const c = Deno.cron("abc", "*/20 * * * *", () => {
+  }, { signal: ac.signal });
+  try {
+    assertThrows(
+      () => Deno.cron("abc", "*/20 * * * *", () => {}),
+      TypeError,
+      "Cron with this name already exists",
+    );
+  } finally {
+    ac.abort();
+    await c;
+  }
+});
+
+Deno.test(async function basicTest() {
+  Deno.env.set("DENO_CRON_TEST_SCHEDULE_OFFSET", "100");
+
+  let count = 0;
+  const promise = deferred();
+  const ac = new AbortController();
+  const c = Deno.cron("abc", "*/20 * * * *", () => {
+    count++;
+    if (count > 5) {
+      promise.resolve();
+    }
+  }, { signal: ac.signal });
+  try {
+    await promise;
+  } finally {
+    ac.abort();
+    await c;
+  }
+});
+
+Deno.test(async function multipleCrons() {
+  Deno.env.set("DENO_CRON_TEST_SCHEDULE_OFFSET", "100");
+
+  let count0 = 0;
+  let count1 = 0;
+  const promise0 = deferred();
+  const promise1 = deferred();
+  const ac = new AbortController();
+  const c0 = Deno.cron("abc", "*/20 * * * *", () => {
+    count0++;
+    if (count0 > 5) {
+      promise0.resolve();
+    }
+  }, { signal: ac.signal });
+  const c1 = Deno.cron("xyz", "*/20 * * * *", () => {
+    count1++;
+    if (count1 > 5) {
+      promise1.resolve();
+    }
+  }, { signal: ac.signal });
+  try {
+    await promise0;
+    await promise1;
+  } finally {
+    ac.abort();
+    await c0;
+    await c1;
+  }
+});
+
+Deno.test(async function overlappingExecutions() {
+  Deno.env.set("DENO_CRON_TEST_SCHEDULE_OFFSET", "100");
+
+  let count = 0;
+  const promise0 = deferred();
+  const promise1 = deferred();
+  const ac = new AbortController();
+  const c = Deno.cron("abc", "*/20 * * * *", async () => {
+    promise0.resolve();
+    count++;
+    await promise1;
+  }, { signal: ac.signal });
+  try {
+    await promise0;
+  } finally {
+    await sleep(2000);
+    promise1.resolve();
+    ac.abort();
+    await c;
+  }
+  assertEquals(count, 1);
+});
+
+Deno.test(async function retriesWithBackkoffSchedule() {
+  Deno.env.set("DENO_CRON_TEST_SCHEDULE_OFFSET", "5000");
+
+  let count = 0;
+  const ac = new AbortController();
+  const c = Deno.cron("abc", "*/20 * * * *", async () => {
+    count += 1;
+    await sleep(10);
+    throw new TypeError("cron error");
+  }, { signal: ac.signal, backoffSchedule: [10, 20] });
+  try {
+    await sleep(6000);
+  } finally {
+    ac.abort();
+    await c;
+  }
+
+  // The cron should have executed 3 times (1st attempt and 2 retries).
+  assertEquals(count, 3);
+});