feat(core): Deno.core.setNextTickCallback (#12771)

This commit adds several new "Deno.core" bindings:
* "setNextTickCallback"
* "hasScheduledTick"
* "setHasScheduledTick"
* "runMicrotasks"
Additionally it changes "Deno.core.setMacrotaskCallback" to
allow registering multiple callbacks. All these changes were necessary
to polyfill "process.nextTick" in Node compat layer.

Co-authored-by: Ben Noordhuis <info@bnoordhuis.nl>

core/bindings.rs

@@ -38,6 +38,18 @@ lazy_static::lazy_static! {
       v8::ExternalReference {
         function: set_macrotask_callback.map_fn_to()
       },
+      v8::ExternalReference {
+        function: set_nexttick_callback.map_fn_to()
+      },
+      v8::ExternalReference {
+        function: run_microtasks.map_fn_to()
+      },
+      v8::ExternalReference {
+        function: has_tick_scheduled.map_fn_to()
+      },
+      v8::ExternalReference {
+        function: set_has_tick_scheduled.map_fn_to()
+      },
       v8::ExternalReference {
         function: eval_context.map_fn_to()
       },
@@ -145,6 +157,20 @@ pub fn initialize_context<'s>(
     "setMacrotaskCallback",
     set_macrotask_callback,
   );
+  set_func(
+    scope,
+    core_val,
+    "setNextTickCallback",
+    set_nexttick_callback,
+  );
+  set_func(scope, core_val, "runMicrotasks", run_microtasks);
+  set_func(scope, core_val, "hasTickScheduled", has_tick_scheduled);
+  set_func(
+    scope,
+    core_val,
+    "setHasTickScheduled",
+    set_has_tick_scheduled,
+  );
   set_func(scope, core_val, "evalContext", eval_context);
   set_func(scope, core_val, "encode", encode);
   set_func(scope, core_val, "decode", decode);
@@ -438,6 +464,51 @@ fn opcall_async<'s>(
   }
 }
 
+fn has_tick_scheduled(
+  scope: &mut v8::HandleScope,
+  _args: v8::FunctionCallbackArguments,
+  mut rv: v8::ReturnValue,
+) {
+  let state_rc = JsRuntime::state(scope);
+  let state = state_rc.borrow();
+  rv.set(to_v8(scope, state.has_tick_scheduled).unwrap());
+}
+
+fn set_has_tick_scheduled(
+  scope: &mut v8::HandleScope,
+  args: v8::FunctionCallbackArguments,
+  _rv: v8::ReturnValue,
+) {
+  let state_rc = JsRuntime::state(scope);
+  let mut state = state_rc.borrow_mut();
+
+  state.has_tick_scheduled = args.get(0).is_true();
+}
+
+fn run_microtasks(
+  scope: &mut v8::HandleScope,
+  _args: v8::FunctionCallbackArguments,
+  _rv: v8::ReturnValue,
+) {
+  scope.perform_microtask_checkpoint();
+}
+
+fn set_nexttick_callback(
+  scope: &mut v8::HandleScope,
+  args: v8::FunctionCallbackArguments,
+  _rv: v8::ReturnValue,
+) {
+  let state_rc = JsRuntime::state(scope);
+  let mut state = state_rc.borrow_mut();
+
+  let cb = match v8::Local::<v8::Function>::try_from(args.get(0)) {
+    Ok(cb) => cb,
+    Err(err) => return throw_type_error(scope, err.to_string()),
+  };
+
+  state.js_nexttick_cbs.push(v8::Global::new(scope, cb));
+}
+
 fn set_macrotask_callback(
   scope: &mut v8::HandleScope,
   args: v8::FunctionCallbackArguments,
@@ -451,17 +522,7 @@ fn set_macrotask_callback(
     Err(err) => return throw_type_error(scope, err.to_string()),
   };
 
-  let slot = match &mut state.js_macrotask_cb {
+  state.js_macrotask_cbs.push(v8::Global::new(scope, cb));
-    slot @ None => slot,
-    _ => {
-      return throw_type_error(
-        scope,
-        "Deno.core.setMacrotaskCallback() already called",
-      );
-    }
-  };
-
-  slot.replace(v8::Global::new(scope, cb));
 }
 
 fn eval_context(

core/lib.deno_core.d.ts

@@ -86,5 +86,26 @@ declare namespace Deno {
     function setWasmStreamingCallback(
       cb: (source: any, rid: number) => void,
     ): void;
+
+    /**
+     * Set a callback that will be called after resolving ops and before resolving
+     * macrotasks.
+     */
+    function setNextTickCallback(
+      cb: () => void,
+    ): void;
+
+    /** Check if there's a scheduled "next tick". */
+    function hasNextTickScheduled(): bool;
+
+    /** Set a value telling the runtime if there are "next ticks" scheduled */
+    function setHasNextTickScheduled(value: bool): void;
+
+    /**
+     * Set a callback that will be called after resolving ops and "next ticks".
+     */
+    function setMacrotaskCallback(
+      cb: () => bool,
+    ): void;
   }
 }

core/runtime.rs

@@ -158,7 +158,9 @@ pub(crate) struct JsRuntimeState {
   pub global_context: Option<v8::Global<v8::Context>>,
   pub(crate) js_recv_cb: Option<v8::Global<v8::Function>>,
   pub(crate) js_sync_cb: Option<v8::Global<v8::Function>>,
-  pub(crate) js_macrotask_cb: Option<v8::Global<v8::Function>>,
+  pub(crate) js_macrotask_cbs: Vec<v8::Global<v8::Function>>,
+  pub(crate) js_nexttick_cbs: Vec<v8::Global<v8::Function>>,
+  pub(crate) has_tick_scheduled: bool,
   pub(crate) js_wasm_streaming_cb: Option<v8::Global<v8::Function>>,
   pub(crate) pending_promise_exceptions:
     HashMap<v8::Global<v8::Promise>, v8::Global<v8::Value>>,
@@ -363,7 +365,9 @@ impl JsRuntime {
       dyn_module_evaluate_idle_counter: 0,
       js_recv_cb: None,
       js_sync_cb: None,
-      js_macrotask_cb: None,
+      js_macrotask_cbs: vec![],
+      js_nexttick_cbs: vec![],
+      has_tick_scheduled: false,
       js_wasm_streaming_cb: None,
       js_error_create_fn,
       pending_ops: FuturesUnordered::new(),
@@ -773,6 +777,7 @@ impl JsRuntime {
     // Ops
     {
       self.resolve_async_ops(cx)?;
+      self.drain_nexttick()?;
       self.drain_macrotasks()?;
       self.check_promise_exceptions()?;
     }
@@ -1549,13 +1554,16 @@ impl JsRuntime {
   }
 
   fn drain_macrotasks(&mut self) -> Result<(), Error> {
-    let js_macrotask_cb_handle =
+    let state = Self::state(self.v8_isolate());
-      match &Self::state(self.v8_isolate()).borrow().js_macrotask_cb {
-        Some(handle) => handle.clone(),
-        None => return Ok(()),
-      };
 
+    if state.borrow().js_macrotask_cbs.is_empty() {
+      return Ok(());
+    }
+
+    let js_macrotask_cb_handles = state.borrow().js_macrotask_cbs.clone();
     let scope = &mut self.handle_scope();
+
+    for js_macrotask_cb_handle in js_macrotask_cb_handles {
       let js_macrotask_cb = js_macrotask_cb_handle.open(scope);
 
       // Repeatedly invoke macrotask callback until it returns true (done),
@@ -1571,7 +1579,7 @@ impl JsRuntime {
         }
 
         if tc_scope.has_terminated() || tc_scope.is_execution_terminating() {
-        break;
+          return Ok(());
         }
 
         let is_done = is_done.unwrap();
@@ -1579,6 +1587,42 @@ impl JsRuntime {
           break;
         }
       }
+    }
+
+    Ok(())
+  }
+
+  fn drain_nexttick(&mut self) -> Result<(), Error> {
+    let state = Self::state(self.v8_isolate());
+
+    if state.borrow().js_nexttick_cbs.is_empty() {
+      return Ok(());
+    }
+
+    if !state.borrow().has_tick_scheduled {
+      let scope = &mut self.handle_scope();
+      scope.perform_microtask_checkpoint();
+    }
+
+    // TODO(bartlomieju): Node also checks for absence of "rejection_to_warn"
+    if !state.borrow().has_tick_scheduled {
+      return Ok(());
+    }
+
+    let js_nexttick_cb_handles = state.borrow().js_nexttick_cbs.clone();
+    let scope = &mut self.handle_scope();
+
+    for js_nexttick_cb_handle in js_nexttick_cb_handles {
+      let js_nexttick_cb = js_nexttick_cb_handle.open(scope);
+
+      let tc_scope = &mut v8::TryCatch::new(scope);
+      let this = v8::undefined(tc_scope).into();
+      js_nexttick_cb.call(tc_scope, this, &[]);
+
+      if let Some(exception) = tc_scope.exception() {
+        return exception_to_err_result(tc_scope, exception, false);
+      }
+    }
 
     Ok(())
   }
@@ -2347,4 +2391,77 @@ assertEquals(1, notify_return_value);
       .unwrap();
     runtime.run_event_loop(false).await.unwrap();
   }
+
+  #[tokio::test]
+  async fn test_set_macrotask_callback_set_next_tick_callback() {
+    async fn op_async_sleep(
+      _op_state: Rc<RefCell<OpState>>,
+      _: (),
+      _: (),
+    ) -> Result<(), Error> {
+      // Future must be Poll::Pending on first call
+      tokio::time::sleep(std::time::Duration::from_millis(1)).await;
+      Ok(())
+    }
+
+    let extension = Extension::builder()
+      .ops(vec![("op_async_sleep", op_async(op_async_sleep))])
+      .build();
+
+    let mut runtime = JsRuntime::new(RuntimeOptions {
+      extensions: vec![extension],
+      ..Default::default()
+    });
+
+    runtime
+      .execute_script(
+        "macrotasks_and_nextticks.js",
+        r#"
+        (async function () {
+          const results = [];
+          Deno.core.setMacrotaskCallback(() => {
+            results.push("macrotask");
+            return true;
+          });
+          Deno.core.setNextTickCallback(() => {
+            results.push("nextTick");
+            Deno.core.setHasTickScheduled(false);
+          });
+
+          Deno.core.setHasTickScheduled(true);
+          await Deno.core.opAsync('op_async_sleep');
+          if (results[0] != "nextTick") {
+            throw new Error(`expected nextTick, got: ${results[0]}`);
+          }
+          if (results[1] != "macrotask") {
+            throw new Error(`expected macrotask, got: ${results[1]}`);
+          }
+        })();
+        "#,
+      )
+      .unwrap();
+    runtime.run_event_loop(false).await.unwrap();
+  }
+
+  #[tokio::test]
+  async fn test_set_macrotask_callback_set_next_tick_callback_multiple() {
+    let mut runtime = JsRuntime::new(Default::default());
+
+    runtime
+      .execute_script(
+        "multiple_macrotasks_and_nextticks.js",
+        r#"
+        Deno.core.setMacrotaskCallback(() => { return true; });
+        Deno.core.setMacrotaskCallback(() => { return true; });
+        Deno.core.setNextTickCallback(() => {});
+        Deno.core.setNextTickCallback(() => {});
+        "#,
+      )
+      .unwrap();
+    let isolate = runtime.v8_isolate();
+    let state_rc = JsRuntime::state(isolate);
+    let state = state_rc.borrow();
+    assert_eq!(state.js_macrotask_cbs.len(), 2);
+    assert_eq!(state.js_nexttick_cbs.len(), 2);
+  }
 }