feat(core): intercept unhandled promise rejections (#12910)

Provide a programmatic means of intercepting rejected promises without a
.catch() handler. Needed for Node compat mode.

Also do a first pass at uncaughtException support because they're
closely intertwined in Node. It's like that Frank Sinatra song:
you can't have one without the other.

Stepping stone for #7013.

3 files changed, 271 insertions, 37 deletions

diff --git a/core/bindings.rs b/core/bindings.rs
index b18d24c79..e4c4e6515 100644
--- a/core/bindings.rs
+++ b/core/bindings.rs
@@ -48,6 +48,12 @@ lazy_static::lazy_static! {
         function: set_nexttick_callback.map_fn_to()
       },
       v8::ExternalReference {
+        function: set_promise_reject_callback.map_fn_to()
+      },
+      v8::ExternalReference {
+        function: set_uncaught_exception_callback.map_fn_to()
+      },
+      v8::ExternalReference {
         function: run_microtasks.map_fn_to()
       },
       v8::ExternalReference {
@@ -171,6 +177,18 @@ pub fn initialize_context<'s>(
     "setNextTickCallback",
     set_nexttick_callback,
   );
+  set_func(
+    scope,
+    core_val,
+    "setPromiseRejectCallback",
+    set_promise_reject_callback,
+  );
+  set_func(
+    scope,
+    core_val,
+    "setUncaughtExceptionCallback",
+    set_uncaught_exception_callback,
+  );
   set_func(scope, core_val, "runMicrotasks", run_microtasks);
   set_func(scope, core_val, "hasTickScheduled", has_tick_scheduled);
   set_func(
@@ -320,30 +338,89 @@ pub extern "C" fn host_initialize_import_meta_object_callback(
 }
 
 pub extern "C" fn promise_reject_callback(message: v8::PromiseRejectMessage) {
+  use v8::PromiseRejectEvent::*;
+
   let scope = &mut unsafe { v8::CallbackScope::new(&message) };
 
   let state_rc = JsRuntime::state(scope);
   let mut state = state_rc.borrow_mut();
 
-  let promise = message.get_promise();
-  let promise_global = v8::Global::new(scope, promise);
+  // Node compat: perform synchronous process.emit("unhandledRejection").
+  //
+  // Note the callback follows the (type, promise, reason) signature of Node's
+  // internal promiseRejectHandler from lib/internal/process/promises.js, not
+  // the (promise, reason) signature of the "unhandledRejection" event listener.
+  //
+  // Short-circuits Deno's regular unhandled rejection logic because that's
+  // a) asynchronous, and b) always terminates.
+  if let Some(js_promise_reject_cb) = state.js_promise_reject_cb.clone() {
+    let js_uncaught_exception_cb = state.js_uncaught_exception_cb.clone();
+    drop(state); // Drop borrow, callbacks can call back into runtime.
+
+    let tc_scope = &mut v8::TryCatch::new(scope);
+    let undefined: v8::Local<v8::Value> = v8::undefined(tc_scope).into();
+    let type_ = v8::Integer::new(tc_scope, message.get_event() as i32);
+    let promise = message.get_promise();
+
+    let reason = match message.get_event() {
+      PromiseRejectWithNoHandler
+      | PromiseRejectAfterResolved
+      | PromiseResolveAfterResolved => message.get_value().unwrap_or(undefined),
+      PromiseHandlerAddedAfterReject => undefined,
+    };
 
-  match message.get_event() {
-    v8::PromiseRejectEvent::PromiseRejectWithNoHandler => {
-      let error = message.get_value().unwrap();
-      let error_global = v8::Global::new(scope, error);
-      state
-        .pending_promise_exceptions
-        .insert(promise_global, error_global);
+    let args = &[type_.into(), promise.into(), reason];
+    js_promise_reject_cb
+      .open(tc_scope)
+      .call(tc_scope, undefined, args);
+
+    if let Some(exception) = tc_scope.exception() {
+      if let Some(js_uncaught_exception_cb) = js_uncaught_exception_cb {
+        tc_scope.reset(); // Cancel pending exception.
+        js_uncaught_exception_cb.open(tc_scope).call(
+          tc_scope,
+          undefined,
+          &[exception],
+        );
+      }
     }
-    v8::PromiseRejectEvent::PromiseHandlerAddedAfterReject => {
-      state.pending_promise_exceptions.remove(&promise_global);
+
+    if tc_scope.has_caught() {
+      // If we get here, an exception was thrown by the unhandledRejection
+      // handler and there is ether no uncaughtException handler or the
+      // handler threw an exception of its own.
+      //
+      // TODO(bnoordhuis) Node terminates the process or worker thread
+      // but we don't really have that option. The exception won't bubble
+      // up either because V8 cancels it when this function returns.
+      let exception = tc_scope
+        .stack_trace()
+        .or_else(|| tc_scope.exception())
+        .map(|value| value.to_rust_string_lossy(tc_scope))
+        .unwrap_or_else(|| "no exception".into());
+      eprintln!("Unhandled exception: {}", exception);
     }
-    v8::PromiseRejectEvent::PromiseRejectAfterResolved => {}
-    v8::PromiseRejectEvent::PromiseResolveAfterResolved => {
-      // Should not warn. See #1272
+  } else {
+    let promise = message.get_promise();
+    let promise_global = v8::Global::new(scope, promise);
+
+    match message.get_event() {
+      PromiseRejectWithNoHandler => {
+        let error = message.get_value().unwrap();
+        let error_global = v8::Global::new(scope, error);
+        state
+          .pending_promise_exceptions
+          .insert(promise_global, error_global);
+      }
+      PromiseHandlerAddedAfterReject => {
+        state.pending_promise_exceptions.remove(&promise_global);
+      }
+      PromiseRejectAfterResolved => {}
+      PromiseResolveAfterResolved => {
+        // Should not warn. See #1272
+      }
     }
-  };
+  }
 }
 
 fn opcall_sync<'s>(
@@ -545,15 +622,12 @@ fn set_nexttick_callback(
   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));
+  if let Ok(cb) = arg0_to_cb(scope, args) {
+    JsRuntime::state(scope)
+      .borrow_mut()
+      .js_nexttick_cbs
+      .push(cb);
+  }
 }
 
 fn set_macrotask_callback(
@@ -561,15 +635,55 @@ fn set_macrotask_callback(
   args: v8::FunctionCallbackArguments,
   _rv: v8::ReturnValue,
 ) {
-  let state_rc = JsRuntime::state(scope);
-  let mut state = state_rc.borrow_mut();
+  if let Ok(cb) = arg0_to_cb(scope, args) {
+    JsRuntime::state(scope)
+      .borrow_mut()
+      .js_macrotask_cbs
+      .push(cb);
+  }
+}
 
-  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()),
-  };
+fn set_promise_reject_callback(
+  scope: &mut v8::HandleScope,
+  args: v8::FunctionCallbackArguments,
+  mut rv: v8::ReturnValue,
+) {
+  if let Ok(new) = arg0_to_cb(scope, args) {
+    if let Some(old) = JsRuntime::state(scope)
+      .borrow_mut()
+      .js_promise_reject_cb
+      .replace(new)
+    {
+      let old = v8::Local::new(scope, old);
+      rv.set(old.into());
+    }
+  }
+}
 
-  state.js_macrotask_cbs.push(v8::Global::new(scope, cb));
+fn set_uncaught_exception_callback(
+  scope: &mut v8::HandleScope,
+  args: v8::FunctionCallbackArguments,
+  mut rv: v8::ReturnValue,
+) {
+  if let Ok(new) = arg0_to_cb(scope, args) {
+    if let Some(old) = JsRuntime::state(scope)
+      .borrow_mut()
+      .js_uncaught_exception_cb
+      .replace(new)
+    {
+      let old = v8::Local::new(scope, old);
+      rv.set(old.into());
+    }
+  }
+}
+
+fn arg0_to_cb(
+  scope: &mut v8::HandleScope,
+  args: v8::FunctionCallbackArguments,
+) -> Result<v8::Global<v8::Function>, ()> {
+  v8::Local::<v8::Function>::try_from(args.get(0))
+    .map(|cb| v8::Global::new(scope, cb))
+    .map_err(|err| throw_type_error(scope, err.to_string()))
 }
 
 fn eval_context(
@@ -707,19 +821,19 @@ fn set_wasm_streaming_callback(
 ) {
   use crate::ops_builtin::WasmStreamingResource;
 
-  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)) {
+  let cb = match arg0_to_cb(scope, args) {
     Ok(cb) => cb,
-    Err(err) => return throw_type_error(scope, err.to_string()),
+    Err(()) => return,
   };
 
+  let state_rc = JsRuntime::state(scope);
+  let mut state = state_rc.borrow_mut();
+
   // The callback to pass to the v8 API has to be a unit type, so it can't
   // borrow or move any local variables. Therefore, we're storing the JS
   // callback in a JsRuntimeState slot.
   if let slot @ None = &mut state.js_wasm_streaming_cb {
-    slot.replace(v8::Global::new(scope, cb));
+    slot.replace(cb);
   } else {
     return throw_type_error(
       scope,
diff --git a/core/lib.deno_core.d.ts b/core/lib.deno_core.d.ts
index 59b2df542..240284c88 100644
--- a/core/lib.deno_core.d.ts
+++ b/core/lib.deno_core.d.ts
@@ -115,5 +115,31 @@ declare namespace Deno {
     function setMacrotaskCallback(
       cb: () => bool,
     ): void;
+
+    /**
+     * Set a callback that will be called when a promise without a .catch
+     * handler is rejected. Returns the old handler or undefined.
+     */
+    function setPromiseRejectCallback(
+      cb: PromiseRejectCallback,
+    ): undefined | PromiseRejectCallback;
+
+    export type PromiseRejectCallback = (
+      type: number,
+      promise: Promise,
+      reason: any,
+    ) => void;
+
+    /**
+     * Set a callback that will be called when an exception isn't caught
+     * by any try/catch handlers. Currently only invoked when the callback
+     * to setPromiseRejectCallback() throws an exception but that is expected
+     * to change in the future. Returns the old handler or undefined.
+     */
+    function setUncaughtExceptionCallback(
+      cb: UncaughtExceptionCallback,
+    ): undefined | UncaughtExceptionCallback;
+
+    export type UncaughtExceptionCallback = (err: any) => void;
   }
 }
diff --git a/core/runtime.rs b/core/runtime.rs
index ad7f16886..6a3d694e6 100644
--- a/core/runtime.rs
+++ b/core/runtime.rs
@@ -144,6 +144,8 @@ pub(crate) struct JsRuntimeState {
   pub(crate) js_sync_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) js_promise_reject_cb: Option<v8::Global<v8::Function>>,
+  pub(crate) js_uncaught_exception_cb: Option<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:
@@ -351,6 +353,8 @@ impl JsRuntime {
       js_sync_cb: None,
       js_macrotask_cbs: vec![],
       js_nexttick_cbs: vec![],
+      js_promise_reject_cb: None,
+      js_uncaught_exception_cb: None,
       has_tick_scheduled: false,
       js_wasm_streaming_cb: None,
       js_error_create_fn,
@@ -2642,4 +2646,94 @@ assertEquals(1, notify_return_value);
       .to_string()
       .contains("JavaScript execution has been terminated"));
   }
+
+  #[tokio::test]
+  async fn test_set_promise_reject_callback() {
+    let promise_reject = Arc::new(AtomicUsize::default());
+    let promise_reject_ = Arc::clone(&promise_reject);
+
+    let uncaught_exception = Arc::new(AtomicUsize::default());
+    let uncaught_exception_ = Arc::clone(&uncaught_exception);
+
+    let op_promise_reject = move |_: &mut OpState, _: (), _: ()| {
+      promise_reject_.fetch_add(1, Ordering::Relaxed);
+      Ok(())
+    };
+
+    let op_uncaught_exception = move |_: &mut OpState, _: (), _: ()| {
+      uncaught_exception_.fetch_add(1, Ordering::Relaxed);
+      Ok(())
+    };
+
+    let extension = Extension::builder()
+      .ops(vec![("op_promise_reject", op_sync(op_promise_reject))])
+      .ops(vec![(
+        "op_uncaught_exception",
+        op_sync(op_uncaught_exception),
+      )])
+      .build();
+
+    let mut runtime = JsRuntime::new(RuntimeOptions {
+      extensions: vec![extension],
+      ..Default::default()
+    });
+
+    runtime
+      .execute_script(
+        "promise_reject_callback.js",
+        r#"
+        // Note: |promise| is not the promise created below, it's a child.
+        Deno.core.setPromiseRejectCallback((type, promise, reason) => {
+          if (type !== /* PromiseRejectWithNoHandler */ 0) {
+            throw Error("unexpected type: " + type);
+          }
+          if (reason.message !== "reject") {
+            throw Error("unexpected reason: " + reason);
+          }
+          Deno.core.opSync("op_promise_reject");
+          throw Error("promiseReject"); // Triggers uncaughtException handler.
+        });
+
+        Deno.core.setUncaughtExceptionCallback((err) => {
+          if (err.message !== "promiseReject") throw err;
+          Deno.core.opSync("op_uncaught_exception");
+        });
+
+        new Promise((_, reject) => reject(Error("reject")));
+        "#,
+      )
+      .unwrap();
+    runtime.run_event_loop(false).await.unwrap();
+
+    assert_eq!(1, promise_reject.load(Ordering::Relaxed));
+    assert_eq!(1, uncaught_exception.load(Ordering::Relaxed));
+
+    runtime
+      .execute_script(
+        "promise_reject_callback.js",
+        r#"
+        {
+          const prev = Deno.core.setPromiseRejectCallback((...args) => {
+            prev(...args);
+          });
+        }
+
+        {
+          const prev = Deno.core.setUncaughtExceptionCallback((...args) => {
+            prev(...args);
+            throw Error("fail");
+          });
+        }
+
+        new Promise((_, reject) => reject(Error("reject")));
+        "#,
+      )
+      .unwrap();
+    // Exception from uncaughtException handler doesn't bubble up but is
+    // printed to stderr.
+    runtime.run_event_loop(false).await.unwrap();
+
+    assert_eq!(2, promise_reject.load(Ordering::Relaxed));
+    assert_eq!(2, uncaught_exception.load(Ordering::Relaxed));
+  }
 }