Repository: denoland/deno_core
Pull request #377 (2023-12-07 15:49:32)

feat(core): Adding deno_core timers API
This is a new fundamental timer API that supports the "run steps after a timeout" specification of https://html.spec.whatwg.org/multipage/timers-and-user-prompts.html


Patch:
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 2ea948d..4f8c4a7 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -74,7 +74,7 @@ jobs:
           # Install and run miri tests
           rustup component add --toolchain nightly miri
           # This somehow prints errors in CI that don't show up locally
-          RUSTFLAGS=-Awarnings cargo +nightly miri test -p deno_core
+          MIRIFLAGS="-Zmiri-disable-isolation" RUSTFLAGS=-Awarnings cargo +nightly miri test -p deno_core
 
       - name: Run examples
         # Only regression tests here for now.
diff --git a/Cargo.lock b/Cargo.lock
index 0a75711..eacd66b 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -267,6 +267,7 @@ dependencies = [
  "deno_ast",
  "deno_ops",
  "deno_unsync",
+ "fastrand",
  "futures",
  "libc",
  "log",
@@ -418,6 +419,12 @@ dependencies = [
  "windows-sys",
 ]
 
+[[package]]
+name = "fastrand"
+version = "2.0.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "25cbce373ec4653f1a01a31e8a5e5ec0c622dc27ff9c4e6606eefef5cbbed4a5"
+
 [[package]]
 name = "form_urlencoded"
 version = "1.2.1"
diff --git a/Cargo.toml b/Cargo.toml
index 2bed68e..58c60c2 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -31,6 +31,7 @@ bencher = "0.1"
 bytes = "1.4.0"
 cooked-waker = "5"
 derive_more = "0.99.17"
+fastrand = "2"
 futures = "0.3.21"
 libc = "0.2.126"
 log = "0.4.17"
diff --git a/core/01_core.js b/core/01_core.js
index c8be033..6f34b6a 100644
--- a/core/01_core.js
+++ b/core/01_core.js
@@ -72,6 +72,7 @@
   registerErrorClass("URIError", URIError);
 
   let unhandledPromiseRejectionHandler = () => false;
+  let timerDepth = 0;
 
   let nextPromiseId = 0;
   const promiseMap = new SafeMap();
@@ -165,7 +166,7 @@
   // responses of async ops.
   function eventLoopTick() {
     // First respond to all pending ops.
-    for (let i = 0; i < arguments.length - 2; i += 2) {
+    for (let i = 0; i < arguments.length - 3; i += 2) {
       const promiseId = arguments[i];
       const res = arguments[i + 1];
       const promise = getPromise(promiseId);
@@ -201,8 +202,17 @@
       }
     }
 
+    const timers = arguments[arguments.length - 2];
+    if (timers) {
+      for (let i = 0; i < timers.length; i += 2) {
+        timerDepth = timers[i];
+        timers[i + 1]();
+      }
+      timerDepth = 0;
+    }
+
     // If we have any rejections for this tick, attempt to process them
-    const rejections = arguments[arguments.length - 2];
+    const rejections = arguments[arguments.length - 3];
     if (rejections) {
       for (let i = 0; i < rejections.length; i += 2) {
         const handled = unhandledPromiseRejectionHandler(
@@ -840,6 +850,12 @@
       unhandledPromiseRejectionHandler = handler,
     reportUnhandledException: (e) => ops.op_dispatch_exception(e, false),
     reportUnhandledPromiseRejection: (e) => ops.op_dispatch_exception(e, true),
+    queueTimer: (depth, repeat, timeout, task) =>
+      ops.op_timer_queue(depth, repeat, timeout, task),
+    cancelTimer: (id) => ops.op_timer_cancel(id),
+    refTimer: (id) => ops.op_timer_ref(id),
+    unrefTimer: (id) => ops.op_timer_unref(id),
+    getTimerDepth: () => timerDepth,
     build,
     setBuildInfo,
     currentUserCallSite,
diff --git a/core/Cargo.toml b/core/Cargo.toml
index 6925106..b5a868e 100644
--- a/core/Cargo.toml
+++ b/core/Cargo.toml
@@ -39,13 +39,14 @@ static_assertions.workspace = true
 tokio.workspace = true
 url.workspace = true
 v8.workspace = true
+cooked-waker.workspace = true
 
 [dev-dependencies]
-cooked-waker.workspace = true
 deno_ast.workspace = true
 bencher.workspace = true
 pretty_assertions.workspace = true
 rstest.workspace = true
+fastrand.workspace = true
 
 [[bench]]
 name = "ops_sync"
diff --git a/core/lib.deno_core.d.ts b/core/lib.deno_core.d.ts
index f9d7e3f..0c64fef 100644
--- a/core/lib.deno_core.d.ts
+++ b/core/lib.deno_core.d.ts
@@ -135,6 +135,26 @@ declare namespace Deno {
     /** Set a value telling the runtime if there are "next ticks" scheduled */
     function setHasNextTickScheduled(value: boolean): void;
 
+    /** Enqueue a timer at the given depth, optionally repeating. */
+    function queueTimer(
+      depth: number,
+      repeat: boolean,
+      delay: number,
+      callback: () => void,
+    ): number;
+
+    /** Cancel a timer with a given ID. */
+    function cancelTimer(id: number);
+
+    /** Ref a timer with a given ID, blocking the runtime from exiting if the timer is still running. */
+    function refTimer(id: number);
+
+    /** Unref a timer with a given ID, allowing the runtime to exit if the timer is still running. */
+    function unrefTimer(id: number);
+
+    /** Gets the current timer depth. */
+    function getTimerDepth(): number;
+
     /**
      * Set a callback that will be called after resolving ops and "next ticks".
      */
diff --git a/core/lib.rs b/core/lib.rs
index e185b7a..b60f78b 100644
--- a/core/lib.rs
+++ b/core/lib.rs
@@ -24,6 +24,7 @@ mod resources;
 mod runtime;
 mod source_map;
 mod tasks;
+mod web_timeout;
 
 // Re-exports
 pub use anyhow;
@@ -198,7 +199,8 @@ macro_rules! located_script_name {
 
 #[cfg(all(test, not(miri)))]
 mod tests {
-  use std::process::{Command, Stdio};
+  use std::process::Command;
+  use std::process::Stdio;
 
   use super::*;
 
diff --git a/core/ops_builtin.rs b/core/ops_builtin.rs
index fdadaa9..4cca5bb 100644
--- a/core/ops_builtin.rs
+++ b/core/ops_builtin.rs
@@ -45,6 +45,10 @@ deno_core::extension!(
     op_format_file_name,
     op_is_proxy,
     op_str_byte_length,
+    ops_builtin_v8::op_timer_queue,
+    ops_builtin_v8::op_timer_cancel,
+    ops_builtin_v8::op_timer_ref,
+    ops_builtin_v8::op_timer_unref,
     ops_builtin_v8::op_ref_op,
     ops_builtin_v8::op_unref_op,
     ops_builtin_v8::op_lazy_load_esm,
diff --git a/core/ops_builtin_v8.rs b/core/ops_builtin_v8.rs
index 7bc4408..3bbb7be 100644
--- a/core/ops_builtin_v8.rs
+++ b/core/ops_builtin_v8.rs
@@ -35,6 +35,46 @@ pub fn op_unref_op(scope: &mut v8::HandleScope, promise_id: i32) {
   context_state.unrefed_ops.borrow_mut().insert(promise_id);
 }
 
+/// Queue a timer, returning a "large" integer in an f64 (allowing up to `MAX_SAFE_INTEGER`
+/// timers to exist).
+#[op2]
+pub fn op_timer_queue(
+  scope: &mut v8::HandleScope,
+  depth: u32,
+  repeat: bool,
+  timeout_ms: f64,
+  #[global] task: v8::Global<v8::Function>,
+) -> f64 {
+  let context_state = JsRealm::state_from_scope(scope);
+  if repeat {
+    context_state
+      .timers
+      .queue_timer_repeat(timeout_ms as _, (task, depth)) as _
+  } else {
+    context_state
+      .timers
+      .queue_timer(timeout_ms as _, (task, depth)) as _
+  }
+}
+
+#[op2]
+pub fn op_timer_cancel(scope: &mut v8::HandleScope, id: f64) {
+  let context_state = JsRealm::state_from_scope(scope);
+  context_state.timers.cancel_timer(id as _);
+}
+
+#[op2]
+pub fn op_timer_ref(scope: &mut v8::HandleScope, id: f64) {
+  let context_state = JsRealm::state_from_scope(scope);
+  context_state.timers.ref_timer(id as _);
+}
+
+#[op2]
+pub fn op_timer_unref(scope: &mut v8::HandleScope, id: f64) {
+  let context_state = JsRealm::state_from_scope(scope);
+  context_state.timers.unref_timer(id as _);
+}
+
 #[op2(reentrant)]
 #[global]
 pub fn op_lazy_load_esm(
diff --git a/core/runtime/jsrealm.rs b/core/runtime/jsrealm.rs
index 168a312..517d8bb 100644
--- a/core/runtime/jsrealm.rs
+++ b/core/runtime/jsrealm.rs
@@ -10,6 +10,7 @@ use crate::modules::ModuleMap;
 use crate::ops::OpCtx;
 use crate::ops::PendingOp;
 use crate::tasks::V8TaskSpawnerFactory;
+use crate::web_timeout::WebTimers;
 use anyhow::Error;
 use deno_unsync::JoinSet;
 use futures::stream::StreamExt;
@@ -47,6 +48,7 @@ impl Hasher for IdentityHasher {
 #[derive(Default)]
 pub(crate) struct ContextState {
   pub(crate) task_spawner_factory: Arc<V8TaskSpawnerFactory>,
+  pub(crate) timers: WebTimers<(v8::Global<v8::Function>, u32)>,
   pub(crate) js_event_loop_tick_cb:
     RefCell<Option<Rc<v8::Global<v8::Function>>>>,
   pub(crate) js_wasm_streaming_cb:
diff --git a/core/runtime/jsruntime.rs b/core/runtime/jsruntime.rs
index 44b2441..4573f15 100644
--- a/core/runtime/jsruntime.rs
+++ b/core/runtime/jsruntime.rs
@@ -1822,6 +1822,7 @@ impl EventLoopPendingState {
     let num_unrefed_ops = state.unrefed_ops.borrow().len();
     let num_pending_ops = state.pending_ops.borrow().len();
     let has_pending_tasks = state.task_spawner_factory.has_pending_tasks();
+    let has_pending_timers = state.timers.has_pending_timers();
     let has_pending_dyn_imports = modules.has_pending_dynamic_imports();
     let has_pending_dyn_module_evaluation =
       modules.has_pending_dyn_module_evaluation();
@@ -1833,6 +1834,7 @@ impl EventLoopPendingState {
       .is_empty();
     EventLoopPendingState {
       has_pending_refed_ops: has_pending_tasks
+        || has_pending_timers
         || num_pending_ops > num_unrefed_ops,
       has_pending_dyn_imports,
       has_pending_dyn_module_evaluation,
@@ -2098,6 +2100,22 @@ impl JsRuntime {
 
     args.push(rejections);
 
+    let timers =
+      if let Poll::Ready(timers) = context_state.timers.poll_timers(cx) {
+        let arr = v8::Array::new(scope, (timers.len() * 2) as _);
+        #[allow(clippy::needless_range_loop)]
+        for i in 0..timers.len() {
+          let value = v8::Integer::new(scope, timers[i].1 .1 as _);
+          arr.set_index(scope, (i * 2) as _, value.into());
+          let value = v8::Local::new(scope, timers[i].1 .0.clone());
+          arr.set_index(scope, (i * 2 + 1) as _, value.into());
+        }
+        arr.into()
+      } else {
+        undefined
+      };
+    args.push(timers);
+
     let has_tick_scheduled = v8::Boolean::new(scope, has_tick_scheduled);
     args.push(has_tick_scheduled.into());
 
diff --git a/core/web_timeout.rs b/core/web_timeout.rs
new file mode 100644
index 0000000..2fb1509
--- /dev/null
+++ b/core/web_timeout.rs
@@ -0,0 +1,635 @@
+// Copyright 2018-2023 the Deno authors. All rights reserved. MIT license.
+use cooked_waker::IntoWaker;
+use cooked_waker::ViaRawPointer;
+use cooked_waker::Wake;
+use cooked_waker::WakeRef;
+use futures::Future;
+use std::cell::Cell;
+use std::cell::RefCell;
+use std::cell::UnsafeCell;
+use std::collections::BTreeMap;
+use std::collections::BTreeSet;
+use std::mem::MaybeUninit;
+use std::num::NonZeroU64;
+use std::pin::Pin;
+use std::task::ready;
+use std::task::Context;
+use std::task::Poll;
+use std::task::Waker;
+use std::time::Duration;
+use tokio::time::Instant;
+use tokio::time::Sleep;
+
+pub(crate) type WebTimerId = u64;
+
+#[derive(PartialEq, Eq, PartialOrd, Ord)]
+enum TimerType {
+  Repeat(NonZeroU64),
+  Once,
+}
+
+#[derive(PartialEq, Eq, PartialOrd, Ord)]
+struct TimerKey(Instant, u64, TimerType);
+
+/// Implements much of the specification described by https://html.spec.whatwg.org/multipage/timers-and-user-prompts.html.
+///
+/// To ensure that we perform well in the face of large numbers of timers, this implementation assumes
+/// that timers may complete in batches that are properly ordered according to the spec. That is to say, timers
+/// are executed in an order according to the following specification:
+///
+/// > Wait until any invocations of this algorithm that had the same global and orderingIdentifier,
+/// > that started before this one, and whose milliseconds is equal to or less than this one's,
+/// > have completed.
+///
+/// This complicates timer resolution because timers fire in an order based on _milliseconds_ (aka their original
+/// timeout duration) rather than an exact moment in time.
+///
+/// While we respect the spirit of this paragraph, we make an assumption that all timer callback invocations are
+/// instantaneous. This means that we can assume that no further timers have elapsed during the execution of a batch of
+/// timers. This does not always hold up in reality (for example, a poorly-written timer in a worker may block on
+/// `Atomics.wait` or a synchronous `XMLHttpRequest`), but it allows us to process timers in a simpler and less-racy way.
+///
+/// We also assume that our underlying timers are high-resolution, and that our underlying timer source resolves in
+/// a proper start time + expiration time order.
+///
+/// Finally, we assume that the event loop time -- the time between which we are able to poll this set of timers -- is
+/// non-zero, and that multiple timers may fire during this period and require re-ordering by their original millisecond
+/// timeouts.
+///
+///
+/// https://github.com/denoland/deno/pull/12953 -- Add refTimer, unrefTimer API
+/// https://github.com/denoland/deno/pull/12862 -- Refactor timers to use one async op per timer
+///
+/// https://github.com/denoland/deno/issues/11398 -- Spurious assertion error when the callback to setInterval lasts longer than the interval
+pub(crate) struct WebTimers<T> {
+  next_id: Cell<WebTimerId>,
+  timers: RefCell<BTreeSet<TimerKey>>,
+  /// We choose a `BTreeMap` over `HashMap` because of memory performance.
+  data_map: RefCell<BTreeMap<WebTimerId, (T, bool)>>,
+  /// How may deleted entries are in the timers BTreeMap? We use this to determine
+  /// when there's too much garbage and need to rebuild the map.
+  tombstone_count: Cell<usize>,
+  /// How many unref'd timers exist?
+  unrefd_count: Cell<usize>,
+  /// A boxed MutableSleep that will allow us to change the Tokio sleep timeout as needed.
+  /// Because this is boxed, we can "safely" unsafely poll it.
+  sleep: Box<MutableSleep>,
+}
+
+impl<T> Default for WebTimers<T> {
+  fn default() -> Self {
+    Self {
+      next_id: Default::default(),
+      timers: Default::default(),
+      data_map: Default::default(),
+      tombstone_count: Default::default(),
+      unrefd_count: Default::default(),
+      sleep: MutableSleep::new(),
+    }
+  }
+}
+
+struct MutableSleep {
+  sleep: UnsafeCell<Option<Sleep>>,
+  ready: Cell<bool>,
+  external_waker: UnsafeCell<Option<Waker>>,
+  internal_waker: Waker,
+}
+
+#[allow(clippy::borrowed_box)]
+impl MutableSleep {
+  fn new() -> Box<Self> {
+    let mut new = Box::new(MaybeUninit::<Self>::uninit());
+
+    new.write(MutableSleep {
+      sleep: Default::default(),
+      ready: Cell::default(),
+      external_waker: UnsafeCell::default(),
+      internal_waker: MutableSleepWaker {
+        inner: new.as_ptr(),
+      }
+      .into_waker(),
+    });
+    unsafe { std::mem::transmute(new) }
+  }
+
+  fn poll_ready(self: &Box<Self>, cx: &mut Context) -> Poll<()> {
+    if self.ready.take() {
+      Poll::Ready(())
+    } else {
+      let external =
+        unsafe { self.external_waker.get().as_mut().unwrap_unchecked() };
+      if let Some(external) = external {
+        // Already have this waker
+        let waker = cx.waker();
+        if !external.will_wake(waker) {
+          *external = waker.clone();
+        }
+        Poll::Pending
+      } else {
+        *external = Some(cx.waker().clone());
+        Poll::Pending
+      }
+    }
+  }
+
+  fn clear(self: &Box<Self>) {
+    unsafe {
+      *self.sleep.get() = None;
+    }
+  }
+
+  fn change(self: &Box<Self>, instant: Instant) {
+    let pin = unsafe {
+      // First replace the current sleep
+      *self.sleep.get() = Some(tokio::time::sleep_until(instant));
+
+      // Then get ourselves a Pin to this
+      Pin::new_unchecked(
+        self
+          .sleep
+          .get()
+          .as_mut()
+          .unwrap_unchecked()
+          .as_mut()
+          .unwrap_unchecked(),
+      )
+    };
+
+    // Register our waker
+    let waker = &self.internal_waker;
+    if pin.poll(&mut Context::from_waker(waker)).is_ready() {
+      self.ready.set(true);
+      self.internal_waker.wake_by_ref();
+    }
+  }
+}
+
+#[repr(transparent)]
+#[derive(Clone)]
+struct MutableSleepWaker {
+  inner: *const MutableSleep,
+}
+
+unsafe impl Send for MutableSleepWaker {}
+unsafe impl Sync for MutableSleepWaker {}
+
+impl WakeRef for MutableSleepWaker {
+  fn wake_by_ref(&self) {
+    unsafe {
+      let this = self.inner.as_ref().unwrap_unchecked();
+      this.ready.set(true);
+      let waker = this.external_waker.get().as_mut().unwrap_unchecked();
+      if let Some(waker) = waker.as_ref() {
+        waker.wake_by_ref();
+      }
+    }
+  }
+}
+
+impl Wake for MutableSleepWaker {
+  fn wake(self) {
+    self.wake_by_ref()
+  }
+}
+
+impl Drop for MutableSleepWaker {
+  fn drop(&mut self) {}
+}
+
+unsafe impl ViaRawPointer for MutableSleepWaker {
+  type Target = ();
+
+  fn into_raw(self) -> *mut () {
+    self.inner as _
+  }
+
+  unsafe fn from_raw(ptr: *mut ()) -> Self {
+    MutableSleepWaker { inner: ptr as _ }
+  }
+}
+
+impl<T: Clone> WebTimers<T> {
+  /// Refs a timer by ID. Invalid IDs are ignored.
+  pub fn ref_timer(&self, id: WebTimerId) {
+    if let Some((_, unrefd)) = self.data_map.borrow_mut().get_mut(&id) {
+      if std::mem::replace(unrefd, false) {
+        self.unrefd_count.set(self.unrefd_count.get() - 1);
+      }
+    }
+  }
+
+  /// Unrefs a timer by ID. Invalid IDs are ignored.
+  pub fn unref_timer(&self, id: WebTimerId) {
+    if let Some((_, unrefd)) = self.data_map.borrow_mut().get_mut(&id) {
+      if !std::mem::replace(unrefd, true) {
+        self.unrefd_count.set(self.unrefd_count.get() + 1);
+      }
+    }
+  }
+
+  /// Queues a timer to be fired in order with the other timers in this set of timers.
+  pub fn queue_timer(&self, timeout_ms: u64, data: T) -> WebTimerId {
+    self.queue_timer_internal(false, timeout_ms, data)
+  }
+
+  /// Queues a timer to be fired in order with the other timers in this set of timers.
+  pub fn queue_timer_repeat(&self, timeout_ms: u64, data: T) -> WebTimerId {
+    self.queue_timer_internal(true, timeout_ms, data)
+  }
+
+  fn queue_timer_internal(
+    &self,
+    repeat: bool,
+    timeout_ms: u64,
+    data: T,
+  ) -> WebTimerId {
+    let id = self.next_id.get() + 1;
+    self.next_id.set(id);
+
+    let mut timers = self.timers.borrow_mut();
+    let deadline = Instant::now()
+      .checked_add(Duration::from_millis(timeout_ms))
+      .unwrap();
+    if let Some(TimerKey(k, ..)) = timers.first() {
+      if &deadline < k {
+        self.sleep.change(deadline);
+      }
+    } else {
+      self.sleep.change(deadline);
+    }
+
+    let timer_type = if repeat {
+      TimerType::Repeat(
+        NonZeroU64::new(timeout_ms).unwrap_or(NonZeroU64::new(1).unwrap()),
+      )
+    } else {
+      TimerType::Once
+    };
+    timers.insert(TimerKey(deadline, id, timer_type));
+
+    let mut data_map = self.data_map.borrow_mut();
+    data_map.insert(id, (data, false));
+    id
+  }
+
+  /// Cancels a pending timer in this set of timers, returning the associated data if a timer
+  /// with the given ID was found.
+  pub fn cancel_timer(&self, timer: u64) -> Option<T> {
+    let mut data_map = self.data_map.borrow_mut();
+    if let Some((data, unrefd)) = data_map.remove(&timer) {
+      if data_map.is_empty() {
+        // When the # of running timers hits zero, clear the timer tree and
+        // tombstone count.
+        self.unrefd_count.set(0);
+        self.timers.borrow_mut().clear();
+        self.tombstone_count.set(0);
+        self.sleep.clear();
+      } else {
+        if unrefd {
+          self.unrefd_count.set(self.unrefd_count.get() - 1);
+        }
+
+        self.tombstone_count.set(self.tombstone_count.get() + 1);
+      }
+      Some(data)
+    } else {
+      None
+    }
+  }
+
+  /// Poll for any timers that have completed.
+  pub fn poll_timers(&self, cx: &mut Context) -> Poll<Vec<(u64, T)>> {
+    ready!(self.sleep.poll_ready(cx));
+    let now = Instant::now();
+    let mut timers = self.timers.borrow_mut();
+    let mut data = self.data_map.borrow_mut();
+    let mut output = vec![];
+
+    let mut split = timers.split_off(&TimerKey(now, 0, TimerType::Once));
+    std::mem::swap(&mut split, &mut timers);
+    for TimerKey(_, id, timer_type) in split {
+      if let TimerType::Repeat(interval) = timer_type {
+        if let Some((data, _)) = data.get(&id) {
+          output.push((id, data.clone()));
+          timers.insert(TimerKey(
+            now
+              .checked_add(Duration::from_millis(interval.into()))
+              .unwrap(),
+            id,
+            timer_type,
+          ));
+        } else {
+          self.tombstone_count.set(self.tombstone_count.get() - 1);
+        }
+      } else if let Some((data, unrefd)) = data.remove(&id) {
+        if unrefd {
+          self.unrefd_count.set(self.unrefd_count.get() - 1);
+        }
+        output.push((id, data));
+      } else {
+        self.tombstone_count.set(self.tombstone_count.get() - 1);
+      }
+    }
+
+    // In-effective poll, run a front-compaction and try again later
+    if output.is_empty() {
+      // We should never have an ineffective poll when the data map is empty, as we check
+      // for this in cancel_timer.
+      debug_assert!(!data.is_empty());
+      while let Some(TimerKey(_, id, _)) = timers.first() {
+        if data.contains_key(id) {
+          break;
+        } else {
+          timers.pop_first();
+        }
+      }
+      if let Some(TimerKey(k, ..)) = timers.first() {
+        self.sleep.change(*k);
+      }
+      return Poll::Pending;
+    }
+
+    if data.is_empty() {
+      // When the # of running timers hits zero, clear the timer tree and
+      // tombstone count.
+      if self.tombstone_count.take() > 0 {
+        timers.clear();
+        self.sleep.clear();
+      }
+    } else {
+      const COMPACTION_MINIMUM: usize = 16;
+      // If we have more tombstones than data, and tombstones are > COMPACTION_MINIMUM, run a compaction
+      if self.tombstone_count.get() > data.len()
+        && self.tombstone_count.get() > COMPACTION_MINIMUM
+      {
+        self.tombstone_count.set(0);
+        timers.retain(|k| data.contains_key(&k.1));
+      }
+      if let Some(TimerKey(k, ..)) = timers.first() {
+        self.sleep.change(*k);
+      }
+    }
+
+    Poll::Ready(output)
+  }
+
+  /// Is this set of timers empty?
+  #[cfg(test)]
+  pub fn is_empty(&self) -> bool {
+    self.data_map.borrow().is_empty()
+  }
+
+  #[cfg(test)]
+  pub fn len(&self) -> usize {
+    self.data_map.borrow().len()
+  }
+
+  #[cfg(test)]
+  pub fn assert_consistent(&self) {
+    if self.data_map.borrow().is_empty() {
+      // If the data map is empty, we should have no timers, no tombstones and no unref'd count
+      assert_eq!(self.timers.borrow().len(), 0);
+      assert_eq!(self.tombstone_count.get(), 0);
+      assert_eq!(self.unrefd_count.get(), 0);
+    } else {
+      assert_eq!(
+        self.timers.borrow().len(),
+        self.data_map.borrow().len() + self.tombstone_count.get()
+      );
+      assert!(self.unrefd_count.get() <= self.data_map.borrow().len());
+    }
+  }
+
+  pub fn has_pending_timers(&self) -> bool {
+    self.timers.borrow().len() > self.unrefd_count.get()
+  }
+}
+
+#[cfg(test)]
+mod tests {
+  use super::*;
+  use futures::future::poll_fn;
+  use rstest::rstest;
+
+  /// Miri is way too slow here on some of the larger tests.
+  const TEN_THOUSAND: u64 = if cfg!(miri) { 100 } else { 10_000 };
+
+  /// Helper function to support miri + rstest. We cannot use I/O in a miri test.
+  fn async_test<F: Future<Output = T>, T>(f: F) -> T {
+    let runtime = tokio::runtime::Builder::new_current_thread()
+      .enable_time()
+      .build()
+      .unwrap();
+    runtime.block_on(f)
+  }
+
+  async fn poll_all(timers: &WebTimers<()>) -> Vec<(u64, ())> {
+    timers.assert_consistent();
+    let len = timers.len();
+    let mut v = vec![];
+    while !timers.is_empty() {
+      let mut batch = poll_fn(|cx| timers.poll_timers(cx)).await;
+      v.append(&mut batch);
+      eprintln!("{}", v.len());
+      timers.assert_consistent();
+    }
+    assert_eq!(v.len(), len);
+    v
+  }
+
+  #[test]
+  fn test_timer() {
+    async_test(async {
+      let timers = WebTimers::<()>::default();
+      let _a = timers.queue_timer(1, ());
+
+      let v = poll_all(&timers).await;
+      assert_eq!(v.len(), 1);
+    });
+  }
+
+  #[rstest]
+  #[test]
+  fn test_timer_cancel_1(#[values(0, 1)] which: u64) {
+    async_test(async {
+      let timers = WebTimers::<()>::default();
+      for i in 0..2 {
+        let id = timers.queue_timer(i * 100, ());
+        if i == which {
+          assert!(timers.cancel_timer(id).is_some());
+        }
+      }
+      assert_eq!(timers.len(), 1);
+
+      let v = poll_all(&timers).await;
+      assert_eq!(v.len(), 1);
+    })
+  }
+
+  #[test]
+  fn test_timers_10_random() {
+    async_test(async {
+      let timers = WebTimers::<()>::default();
+      for i in 0..10 {
+        timers.queue_timer((i % 3) * 10, ());
+      }
+
+      let v = poll_all(&timers).await;
+      assert_eq!(v.len(), 10);
+    })
+  }
+
+  #[test]
+  fn test_timers_10_random_cancel() {
+    async_test(async {
+      let timers = WebTimers::<()>::default();
+      for i in 0..10 {
+        let id = timers.queue_timer((i % 3) * 10, ());
+        timers.cancel_timer(id);
+      }
+
+      let v = poll_all(&timers).await;
+      assert_eq!(v.len(), 0);
+    });
+  }
+
+  #[rstest]
+  #[test]
+  fn test_timers_10_random_cancel_after(#[values(true, false)] reverse: bool) {
+    async_test(async {
+      let timers = WebTimers::<()>::default();
+      let mut ids = vec![];
+      for i in 0..2 {
+        ids.push(timers.queue_timer((i % 3) * 10, ()));
+      }
+      if reverse {
+        ids.reverse();
+      }
+      for id in ids {
+        timers.cancel_timer(id);
+        timers.assert_consistent();
+      }
+
+      let v = poll_all(&timers).await;
+      assert_eq!(v.len(), 0);
+    });
+  }
+
+  #[test]
+  fn test_timers_10() {
+    async_test(async {
+      let timers = WebTimers::<()>::default();
+      for _i in 0..10 {
+        timers.queue_timer(1, ());
+      }
+
+      let v = poll_all(&timers).await;
+      assert_eq!(v.len(), 10);
+    });
+  }
+
+  #[test]
+  fn test_timers_10_000_random() {
+    async_test(async {
+      let timers = WebTimers::<()>::default();
+      for i in 0..TEN_THOUSAND {
+        timers.queue_timer(i % 10, ());
+      }
+
+      let v = poll_all(&timers).await;
+      assert_eq!(v.len(), TEN_THOUSAND as usize);
+    });
+  }
+
+  /// Cancel a large number of timers at the head of the queue to trigger
+  /// a front compaction.
+  #[test]
+  fn test_timers_cancel_first() {
+    async_test(async {
+      let timers = WebTimers::<()>::default();
+      let mut ids = vec![];
+      for _ in 0..TEN_THOUSAND {
+        ids.push(timers.queue_timer(1, ()));
+      }
+      for _ in 0..10 {
+        timers.queue_timer(1000, ());
+      }
+      for id in ids {
+        timers.cancel_timer(id);
+      }
+      let v = poll_all(&timers).await;
+      assert_eq!(v.len(), 10);
+    });
+  }
+
+  #[test]
+  fn test_timers_10_000_cancel_most() {
+    async_test(async {
+      let timers = WebTimers::<()>::default();
+      let mut ids = vec![];
+      for i in 0..TEN_THOUSAND {
+        ids.push(timers.queue_timer(i % 100, ()));
+      }
+
+      // This should trigger a compaction
+      fastrand::seed(42);
+      ids.retain(|_| fastrand::u8(0..10) > 0);
+      for id in ids.iter() {
+        timers.cancel_timer(*id);
+      }
+
+      let v = poll_all(&timers).await;
+      assert_eq!(v.len(), TEN_THOUSAND as usize - ids.len());
+    });
+  }
+
+  #[rstest]
+  #[test]
+  fn test_chaos(#[values(42, 99, 1000)] seed: u64) {
+    async_test(async {
+      let timers = WebTimers::<()>::default();
+      fastrand::seed(seed);
+
+      let mut count = 0;
+      let mut ref_count = 0;
+
+      for _ in 0..TEN_THOUSAND {
+        let mut cancelled = false;
+        let mut unrefed = false;
+        let id = timers.queue_timer(fastrand::u64(0..10), ());
+        for _ in 0..fastrand::u64(0..10) {
+          if fastrand::u8(0..10) == 0 {
+            timers.cancel_timer(id);
+            cancelled = true;
+          }
+          if fastrand::u8(0..10) == 0 {
+            timers.ref_timer(id);
+            unrefed = false;
+          }
+          if fastrand::u8(0..10) == 0 {
+            timers.unref_timer(id);
+            unrefed = true;
+          }
+        }
+
+        if !cancelled {
+          count += 1;
+        }
+        if !unrefed {
+          ref_count += 1;
+        }
+
+        timers.assert_consistent();
+      }
+
+      eprintln!("count={count} ref_count={ref_count}");
+
+      let v = poll_all(&timers).await;
+      assert_eq!(v.len(), count);
+
+      assert!(timers.is_empty());
+      assert!(!timers.has_pending_timers());
+    });
+  }
+}
diff --git a/ops/op2/config.rs b/ops/op2/config.rs
index ab0bbad..2f3b253 100644
--- a/ops/op2/config.rs
+++ b/ops/op2/config.rs
@@ -170,7 +170,8 @@ impl MacroConfig {
 #[cfg(test)]
 mod tests {
   use super::*;
-  use syn::{ItemFn, Meta};
+  use syn::ItemFn;
+  use syn::Meta;
 
   fn test_parse(s: &str, expected: MacroConfig) {
     let item_fn = syn::parse_str::<ItemFn>(&format!("#[op2{s}] fn x() {{ }}"))


Test patch:
diff --git a/testing/checkin.d.ts b/testing/checkin.d.ts
index fdb754f..28d144d 100644
--- a/testing/checkin.d.ts
+++ b/testing/checkin.d.ts
@@ -1,3 +1,13 @@
 // Copyright 2018-2023 the Deno authors. All rights reserved. MIT license.
 /// <reference path="../core/lib.deno_core.d.ts" />
 /// <reference lib="dom" />
+
+// Types and method unavailable in TypeScript by default.
+interface PromiseConstructor {
+  withResolvers<T>(): {
+    promise: Promise<T>;
+    resolve: (value: T | PromiseLike<T>) => void;
+    // deno-lint-ignore no-explicit-any
+    reject: (reason?: any) => void;
+  };
+}
diff --git a/testing/checkin/runner/mod.rs b/testing/checkin/runner/mod.rs
index da17e06..25af0ab 100644
--- a/testing/checkin/runner/mod.rs
+++ b/testing/checkin/runner/mod.rs
@@ -33,6 +33,7 @@ deno_core::extension!(
     "__init.js",
     "console.ts" with_specifier "checkin:console",
     "testing.ts" with_specifier "checkin:testing",
+    "timers.ts" with_specifier "checkin:timers",
   ],
   state = |state| {
     state.put(TestFunctions::default());
diff --git a/testing/checkin/runtime/__init.js b/testing/checkin/runtime/__init.js
index 0b5b371..acae428 100644
--- a/testing/checkin/runtime/__init.js
+++ b/testing/checkin/runtime/__init.js
@@ -1,7 +1,12 @@
 // Copyright 2018-2023 the Deno authors. All rights reserved. MIT license.
 import * as testing from "checkin:testing";
 import * as console from "checkin:console";
+import * as timers from "checkin:timers";
 
 testing;
 
 globalThis.console = console.console;
+globalThis.setTimeout = timers.setTimeout;
+globalThis.setInterval = timers.setInterval;
+globalThis.clearTimeout = timers.clearTimeout;
+globalThis.clearInterval = timers.clearInterval;
diff --git a/testing/checkin/runtime/timers.ts b/testing/checkin/runtime/timers.ts
new file mode 100644
index 0000000..c7b4be1
--- /dev/null
+++ b/testing/checkin/runtime/timers.ts
@@ -0,0 +1,39 @@
+// Copyright 2018-2023 the Deno authors. All rights reserved. MIT license.
+
+/**
+ * Call a callback function after a delay.
+ */
+export function setTimeout(callback, delay) {
+  return Deno.core.queueTimer(
+    Deno.core.getTimerDepth() + 1,
+    false,
+    delay,
+    callback,
+  );
+}
+
+/**
+ * Call a callback function after a delay.
+ */
+export function setInterval(callback, delay) {
+  return Deno.core.queueTimer(
+    Deno.core.getTimerDepth() + 1,
+    true,
+    delay,
+    callback,
+  );
+}
+
+/**
+ * Clear a timeout or interval.
+ */
+export function clearTimeout(id) {
+  Deno.core.cancelTimer(id);
+}
+
+/**
+ * Clear a timeout or interval.
+ */
+export function clearInterval(id) {
+  Deno.core.cancelTimer(id);
+}
diff --git a/testing/integration/timer_many/timer_many.out b/testing/integration/timer_many/timer_many.out
new file mode 100644
index 0000000..749fce6
--- /dev/null
+++ b/testing/integration/timer_many/timer_many.out
@@ -0,0 +1,1 @@
+1000000
diff --git a/testing/integration/timer_many/timer_many.ts b/testing/integration/timer_many/timer_many.ts
new file mode 100644
index 0000000..af5e70b
--- /dev/null
+++ b/testing/integration/timer_many/timer_many.ts
@@ -0,0 +1,4 @@
+// Copyright 2018-2023 the Deno authors. All rights reserved. MIT license.
+let n = 0;
+for (let i = 0; i < 1e6; i++) setTimeout(() => n++, 1);
+setTimeout(() => console.log(n), 2);
diff --git a/testing/integration/timer_ref/timer_ref.out b/testing/integration/timer_ref/timer_ref.out
new file mode 100644
index 0000000..422c2b7
--- /dev/null
+++ b/testing/integration/timer_ref/timer_ref.out
@@ -0,0 +1,2 @@
+a
+b
diff --git a/testing/integration/timer_ref/timer_ref.ts b/testing/integration/timer_ref/timer_ref.ts
new file mode 100644
index 0000000..42eec7b
--- /dev/null
+++ b/testing/integration/timer_ref/timer_ref.ts
@@ -0,0 +1,5 @@
+// Copyright 2018-2023 the Deno authors. All rights reserved. MIT license.
+setTimeout(() => console.log("a"), 1000);
+setTimeout(() => console.log("b"), 2000);
+const c = setTimeout(() => console.log("c"), 3000);
+Deno.core.unrefTimer(c);
diff --git a/testing/lib.rs b/testing/lib.rs
index 3083790..c165077 100644
--- a/testing/lib.rs
+++ b/testing/lib.rs
@@ -37,7 +37,7 @@ unit_test!(
   microtask_test,
   serialize_deserialize_test,
   tc39_test,
-  test_test,
+  timer_test,
 );
 
 // Test the load and run of an entire file within the `checkin` infrastructure.
@@ -49,4 +49,6 @@ integration_test!(
   error_with_stack,
   error_without_stack,
   smoke_test,
+  timer_ref,
+  timer_many,
 );
diff --git a/testing/unit/timer_test.ts b/testing/unit/timer_test.ts
new file mode 100644
index 0000000..5eb353e
--- /dev/null
+++ b/testing/unit/timer_test.ts
@@ -0,0 +1,64 @@
+// Copyright 2018-2023 the Deno authors. All rights reserved. MIT license.
+import { assert, assertEquals, test } from "checkin:testing";
+
+test(async function testTimeout() {
+  const { promise, resolve } = Promise.withResolvers();
+  const id = setTimeout(() => {
+    resolve(null);
+  }, 1);
+  assert(id > 0);
+  await promise;
+});
+
+test(async function testTimeoutWithPromise() {
+  await new Promise((r) => setTimeout(r, 1));
+});
+
+test(async function testManyTimers() {
+  let n = 0;
+  for (let i = 0; i < 1e6; i++) {
+    setTimeout(() => n++, 1);
+  }
+  await new Promise((r) => setTimeout(r, 2));
+  assertEquals(n, 1e6);
+});
+
+test(async function testManyIntervals() {
+  const expected = 1000;
+  let n = 0;
+  for (let i = 0; i < 100; i++) {
+    let count = 0;
+    const id = setInterval(() => {
+      if (count++ == 10) {
+        clearInterval(id);
+      } else {
+        n++;
+      }
+      assert(n <= expected, `${n} <= ${expected}`);
+    }, 1);
+  }
+  let id: number;
+  await new Promise((r) =>
+    id = setInterval(() => {
+      assert(n <= expected, `${n} <= ${expected}`);
+      if (n == expected) {
+        clearInterval(id);
+        r(null);
+      }
+    }, 1)
+  );
+  assertEquals(n, expected);
+});
+
+test(async function testTimerDepth() {
+  const { promise, resolve } = Promise.withResolvers();
+  assertEquals(Deno.core.getTimerDepth(), 0);
+  setTimeout(() => {
+    assertEquals(Deno.core.getTimerDepth(), 1);
+    setTimeout(() => {
+      assertEquals(Deno.core.getTimerDepth(), 2);
+      resolve(null);
+    }, 1);
+  }, 1);
+  await promise;
+});