From 01d3e0f317ca180bbf0ac8a17c6651869110e02f Mon Sep 17 00:00:00 2001 From: Igor Zinkovsky Date: Wed, 1 Nov 2023 11:57:55 -0700 Subject: 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`. --- ext/cron/01_cron.ts | 58 +++++++++ ext/cron/Cargo.toml | 23 ++++ ext/cron/interface.rs | 23 ++++ ext/cron/lib.rs | 128 +++++++++++++++++++ ext/cron/local.rs | 343 ++++++++++++++++++++++++++++++++++++++++++++++++++ ext/cron/time.rs | 19 +++ 6 files changed, 594 insertions(+) create mode 100644 ext/cron/01_cron.ts create mode 100644 ext/cron/Cargo.toml create mode 100644 ext/cron/interface.rs create mode 100644 ext/cron/lib.rs create mode 100644 ext/cron/local.rs create mode 100644 ext/cron/time.rs (limited to 'ext/cron') diff --git a/ext/cron/01_cron.ts b/ext/cron/01_cron.ts new file mode 100644 index 000000000..a615ae34b --- /dev/null +++ b/ext/cron/01_cron.ts @@ -0,0 +1,58 @@ +// Copyright 2018-2023 the Deno authors. All rights reserved. MIT license. + +// @ts-ignore internal api +const core = Deno.core; + +function cron( + name: string, + schedule: string, + handler: () => Promise | void, + options?: { backoffSchedule?: number[]; signal?: AbortSignal }, +) { + if (name === undefined) { + throw new TypeError("Deno.cron requires a unique name"); + } + if (schedule === undefined) { + throw new TypeError("Deno.cron requires a valid schedule"); + } + if (handler === undefined) { + throw new TypeError("Deno.cron requires a handler"); + } + + const rid = core.ops.op_cron_create( + name, + schedule, + options?.backoffSchedule, + ); + + if (options?.signal) { + const signal = options?.signal; + signal.addEventListener( + "abort", + () => { + core.close(rid); + }, + { once: true }, + ); + } + + return (async () => { + let success = true; + while (true) { + const r = await core.opAsync("op_cron_next", rid, success); + if (r === false) { + break; + } + try { + const result = handler(); + const _res = result instanceof Promise ? (await result) : result; + success = true; + } catch (error) { + console.error(`Exception in cron handler ${name}`, error); + success = false; + } + } + })(); +} + +export { cron }; diff --git a/ext/cron/Cargo.toml b/ext/cron/Cargo.toml new file mode 100644 index 000000000..ac107c567 --- /dev/null +++ b/ext/cron/Cargo.toml @@ -0,0 +1,23 @@ +# Copyright 2018-2023 the Deno authors. All rights reserved. MIT license. + +[package] +name = "deno_cron" +version = "0.1.0" +authors.workspace = true +edition.workspace = true +license.workspace = true +readme = "README.md" +repository.workspace = true +description = "Implementation of the Deno cron API" + +[lib] +path = "lib.rs" + +[dependencies] +anyhow.workspace = true +async-trait.workspace = true +chrono.workspace = true +deno_core.workspace = true +deno_unsync = "0.1.1" +saffron.workspace = true +tokio.workspace = true diff --git a/ext/cron/interface.rs b/ext/cron/interface.rs new file mode 100644 index 000000000..c70988788 --- /dev/null +++ b/ext/cron/interface.rs @@ -0,0 +1,23 @@ +// Copyright 2018-2023 the Deno authors. All rights reserved. MIT license. + +use async_trait::async_trait; +use deno_core::error::AnyError; + +pub trait CronHandler { + type EH: CronHandle + 'static; + + fn create(&self, spec: CronSpec) -> Result; +} + +#[async_trait(?Send)] +pub trait CronHandle { + async fn next(&self, prev_success: bool) -> Result; + fn close(&self); +} + +#[derive(Clone)] +pub struct CronSpec { + pub name: String, + pub cron_schedule: String, + pub backoff_schedule: Option>, +} diff --git a/ext/cron/lib.rs b/ext/cron/lib.rs new file mode 100644 index 000000000..c49659703 --- /dev/null +++ b/ext/cron/lib.rs @@ -0,0 +1,128 @@ +// Copyright 2018-2023 the Deno authors. All rights reserved. MIT license. + +mod interface; +pub mod local; +mod time; + +use std::borrow::Cow; +use std::cell::RefCell; +use std::rc::Rc; + +use deno_core::error::get_custom_error_class; +use deno_core::error::type_error; +use deno_core::error::AnyError; +use deno_core::op2; +use deno_core::OpState; +use deno_core::Resource; +use deno_core::ResourceId; + +pub use crate::interface::*; + +pub const UNSTABLE_FEATURE_NAME: &str = "cron"; + +deno_core::extension!(deno_cron, + deps = [ deno_console ], + parameters = [ C: CronHandler ], + ops = [ + op_cron_create, + op_cron_next, + ], + esm = [ "01_cron.ts" ], + options = { + cron_handler: C, + }, + state = |state, options| { + state.put(Rc::new(options.cron_handler)); + } +); + +struct CronResource { + handle: Rc, +} + +impl Resource for CronResource { + fn name(&self) -> Cow { + "cron".into() + } + + fn close(self: Rc) { + self.handle.close(); + } +} + +#[op2] +#[smi] +fn op_cron_create( + state: Rc>, + #[string] name: String, + #[string] cron_schedule: String, + #[serde] backoff_schedule: Option>, +) -> Result +where + C: CronHandler + 'static, +{ + let cron_handler = { + let state = state.borrow(); + // TODO(bartlomieju): replace with `state.feature_checker.check_or_exit` + // once we phase out `check_or_exit_with_legacy_fallback` + state + .feature_checker + .check_or_exit_with_legacy_fallback(UNSTABLE_FEATURE_NAME, "Deno.cron"); + state.borrow::>().clone() + }; + + validate_cron_name(&name)?; + + let handle = cron_handler.create(CronSpec { + name, + cron_schedule, + backoff_schedule, + })?; + + let handle_rid = { + let mut state = state.borrow_mut(); + state.resource_table.add(CronResource { + handle: Rc::new(handle), + }) + }; + Ok(handle_rid) +} + +#[op2(async)] +async fn op_cron_next( + state: Rc>, + #[smi] rid: ResourceId, + prev_success: bool, +) -> Result +where + C: CronHandler + 'static, +{ + let cron_handler = { + let state = state.borrow(); + let resource = match state.resource_table.get::>(rid) { + Ok(resource) => resource, + Err(err) => { + if get_custom_error_class(&err) == Some("BadResource") { + return Ok(false); + } else { + return Err(err); + } + } + }; + resource.handle.clone() + }; + + cron_handler.next(prev_success).await +} + +fn validate_cron_name(name: &str) -> Result<(), AnyError> { + if name.len() > 64 { + return Err(type_error("Cron name is too long")); + } + if !name.chars().all(|c| { + c.is_ascii_whitespace() || c.is_ascii_alphanumeric() || c == '_' || c == '-' + }) { + return Err(type_error("Invalid cron name")); + } + Ok(()) +} diff --git a/ext/cron/local.rs b/ext/cron/local.rs new file mode 100644 index 000000000..0b6dcae2e --- /dev/null +++ b/ext/cron/local.rs @@ -0,0 +1,343 @@ +// Copyright 2018-2023 the Deno authors. All rights reserved. MIT license. + +use std::cell::OnceCell; +use std::cell::RefCell; +use std::collections::BTreeMap; +use std::collections::HashMap; +use std::env; +use std::rc::Rc; +use std::rc::Weak; +use std::sync::Arc; + +use async_trait::async_trait; +use deno_core::error::type_error; +use deno_core::error::AnyError; +use deno_core::futures; +use deno_core::futures::FutureExt; +use deno_unsync::spawn; +use deno_unsync::JoinHandle; +use tokio::sync::mpsc; +use tokio::sync::mpsc::WeakSender; +use tokio::sync::OwnedSemaphorePermit; +use tokio::sync::Semaphore; + +use crate::CronHandle; +use crate::CronHandler; +use crate::CronSpec; + +const MAX_CRONS: usize = 100; +const DISPATCH_CONCURRENCY_LIMIT: usize = 50; +const MAX_BACKOFF_MS: u32 = 60 * 60 * 1_000; // 1 hour +const MAX_BACKOFF_COUNT: usize = 5; +const DEFAULT_BACKOFF_SCHEDULE: [u32; 5] = [100, 1_000, 5_000, 30_000, 60_000]; + +pub struct LocalCronHandler { + cron_schedule_tx: OnceCell>, + concurrency_limiter: Arc, + cron_loop_join_handle: OnceCell>, + runtime_state: Rc>, +} + +struct RuntimeState { + crons: HashMap, + scheduled_deadlines: BTreeMap>, +} + +struct Cron { + spec: CronSpec, + next_tx: mpsc::WeakSender<()>, + current_execution_retries: u32, +} + +impl Cron { + fn backoff_schedule(&self) -> &[u32] { + self + .spec + .backoff_schedule + .as_deref() + .unwrap_or(&DEFAULT_BACKOFF_SCHEDULE) + } +} + +impl Default for LocalCronHandler { + fn default() -> Self { + Self::new() + } +} + +impl LocalCronHandler { + pub fn new() -> Self { + Self { + cron_schedule_tx: OnceCell::new(), + concurrency_limiter: Arc::new(Semaphore::new(DISPATCH_CONCURRENCY_LIMIT)), + cron_loop_join_handle: OnceCell::new(), + runtime_state: Rc::new(RefCell::new(RuntimeState { + crons: HashMap::new(), + scheduled_deadlines: BTreeMap::new(), + })), + } + } + + async fn cron_loop( + runtime_state: Rc>, + mut cron_schedule_rx: mpsc::Receiver<(String, bool)>, + ) -> Result<(), AnyError> { + loop { + let earliest_deadline = runtime_state + .borrow() + .scheduled_deadlines + .keys() + .next() + .copied(); + + let sleep_fut = if let Some(earliest_deadline) = earliest_deadline { + let now = crate::time::utc_now().timestamp_millis() as u64; + if let Some(delta) = earliest_deadline.checked_sub(now) { + tokio::time::sleep(std::time::Duration::from_millis(delta)).boxed() + } else { + futures::future::ready(()).boxed() + } + } else { + futures::future::pending().boxed() + }; + + let cron_to_schedule = tokio::select! { + _ = sleep_fut => None, + x = cron_schedule_rx.recv() => { + if x.is_none() { + return Ok(()); + }; + x + } + }; + + // Schedule next execution of the cron if needed. + if let Some((name, prev_success)) = cron_to_schedule { + let mut runtime_state = runtime_state.borrow_mut(); + if let Some(cron) = runtime_state.crons.get_mut(&name) { + let backoff_schedule = cron.backoff_schedule(); + let next_deadline = if !prev_success + && cron.current_execution_retries < backoff_schedule.len() as u32 + { + let backoff_ms = + backoff_schedule[cron.current_execution_retries as usize]; + let now = crate::time::utc_now().timestamp_millis() as u64; + cron.current_execution_retries += 1; + now + backoff_ms as u64 + } else { + let next_ts = compute_next_deadline(&cron.spec.cron_schedule)?; + cron.current_execution_retries = 0; + next_ts + }; + runtime_state + .scheduled_deadlines + .entry(next_deadline) + .or_default() + .push(name.to_string()); + } + } + + // Dispatch ready to execute crons. + let crons_to_execute = { + let mut runtime_state = runtime_state.borrow_mut(); + runtime_state.get_ready_crons()? + }; + for (_, tx) in crons_to_execute { + if let Some(tx) = tx.upgrade() { + let _ = tx.send(()).await; + } + } + } + } +} + +impl RuntimeState { + fn get_ready_crons( + &mut self, + ) -> Result)>, AnyError> { + let now = crate::time::utc_now().timestamp_millis() as u64; + + let ready = { + let to_remove = self + .scheduled_deadlines + .range(..=now) + .map(|(ts, _)| *ts) + .collect::>(); + to_remove + .iter() + .flat_map(|ts| { + self + .scheduled_deadlines + .remove(ts) + .unwrap() + .iter() + .map(move |name| (*ts, name.clone())) + .collect::>() + }) + .map(|(_, name)| { + (name.clone(), self.crons.get(&name).unwrap().next_tx.clone()) + }) + .collect::>() + }; + + Ok(ready) + } +} + +#[async_trait(?Send)] +impl CronHandler for LocalCronHandler { + type EH = CronExecutionHandle; + + fn create(&self, spec: CronSpec) -> Result { + // Ensure that the cron loop is started. + self.cron_loop_join_handle.get_or_init(|| { + let (cron_schedule_tx, cron_schedule_rx) = + mpsc::channel::<(String, bool)>(1); + self.cron_schedule_tx.set(cron_schedule_tx).unwrap(); + let runtime_state = self.runtime_state.clone(); + spawn(async move { + LocalCronHandler::cron_loop(runtime_state, cron_schedule_rx) + .await + .unwrap(); + }) + }); + + let mut runtime_state = self.runtime_state.borrow_mut(); + + if runtime_state.crons.len() > MAX_CRONS { + return Err(type_error("Too many crons")); + } + if runtime_state.crons.contains_key(&spec.name) { + return Err(type_error("Cron with this name already exists")); + } + + // Validate schedule expression. + spec + .cron_schedule + .parse::() + .map_err(|_| type_error("Invalid cron schedule"))?; + + // Validate backoff_schedule. + if let Some(backoff_schedule) = &spec.backoff_schedule { + validate_backoff_schedule(backoff_schedule)?; + } + + let (next_tx, next_rx) = mpsc::channel::<()>(1); + let cron = Cron { + spec: spec.clone(), + next_tx: next_tx.downgrade(), + current_execution_retries: 0, + }; + runtime_state.crons.insert(spec.name.clone(), cron); + + Ok(CronExecutionHandle { + name: spec.name.clone(), + cron_schedule_tx: self.cron_schedule_tx.get().unwrap().clone(), + concurrency_limiter: self.concurrency_limiter.clone(), + runtime_state: Rc::downgrade(&self.runtime_state), + inner: RefCell::new(Inner { + next_rx: Some(next_rx), + shutdown_tx: Some(next_tx), + permit: None, + }), + }) + } +} + +pub struct CronExecutionHandle { + name: String, + runtime_state: Weak>, + cron_schedule_tx: mpsc::Sender<(String, bool)>, + concurrency_limiter: Arc, + inner: RefCell, +} + +struct Inner { + next_rx: Option>, + shutdown_tx: Option>, + permit: Option, +} + +#[async_trait(?Send)] +impl CronHandle for CronExecutionHandle { + async fn next(&self, prev_success: bool) -> Result { + self.inner.borrow_mut().permit.take(); + + if self + .cron_schedule_tx + .send((self.name.clone(), prev_success)) + .await + .is_err() + { + return Ok(false); + }; + + let Some(mut next_rx) = self.inner.borrow_mut().next_rx.take() else { + return Ok(false); + }; + if next_rx.recv().await.is_none() { + return Ok(false); + }; + + let permit = self.concurrency_limiter.clone().acquire_owned().await?; + let mut inner = self.inner.borrow_mut(); + inner.next_rx = Some(next_rx); + inner.permit = Some(permit); + Ok(true) + } + + fn close(&self) { + if let Some(tx) = self.inner.borrow_mut().shutdown_tx.take() { + drop(tx) + } + if let Some(runtime_state) = self.runtime_state.upgrade() { + let mut runtime_state = runtime_state.borrow_mut(); + runtime_state.crons.remove(&self.name); + } + } +} + +fn compute_next_deadline(cron_expression: &str) -> Result { + let now = crate::time::utc_now(); + + if let Ok(test_schedule) = env::var("DENO_CRON_TEST_SCHEDULE_OFFSET") { + if let Ok(offset) = test_schedule.parse::() { + return Ok(now.timestamp_millis() as u64 + offset); + } + } + + let cron = cron_expression + .parse::() + .map_err(|_| anyhow::anyhow!("invalid cron expression"))?; + let Some(next_deadline) = cron.next_after(now) else { + return Err(anyhow::anyhow!("invalid cron expression")); + }; + Ok(next_deadline.timestamp_millis() as u64) +} + +fn validate_backoff_schedule( + backoff_schedule: &Vec, +) -> Result<(), AnyError> { + if backoff_schedule.len() > MAX_BACKOFF_COUNT { + return Err(type_error("Invalid backoff schedule")); + } + if backoff_schedule.iter().any(|s| *s > MAX_BACKOFF_MS) { + return Err(type_error("Invalid backoff schedule")); + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_compute_next_deadline() { + let now = crate::time::utc_now().timestamp_millis() as u64; + assert!(compute_next_deadline("*/1 * * * *").unwrap() > now); + assert!(compute_next_deadline("* * * * *").unwrap() > now); + assert!(compute_next_deadline("bogus").is_err()); + assert!(compute_next_deadline("* * * * * *").is_err()); + assert!(compute_next_deadline("* * *").is_err()); + } +} diff --git a/ext/cron/time.rs b/ext/cron/time.rs new file mode 100644 index 000000000..60375818b --- /dev/null +++ b/ext/cron/time.rs @@ -0,0 +1,19 @@ +// Copyright 2018-2023 the Deno authors. All rights reserved. MIT license. + +/// Identical to chrono::Utc::now() but without the system "clock" +/// feature flag. +/// +/// The "clock" feature flag pulls in the "iana-time-zone" crate +/// which links to macOS's "CoreFoundation" framework which increases +/// startup time for the CLI. +pub fn utc_now() -> chrono::DateTime { + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .expect("system time before Unix epoch"); + let naive = chrono::NaiveDateTime::from_timestamp_opt( + now.as_secs() as i64, + now.subsec_nanos(), + ) + .unwrap(); + chrono::DateTime::from_naive_utc_and_offset(naive, chrono::Utc) +} -- cgit v1.2.3