fix(fetch): Fix uncaught rejection panic with `WebAssembly.instantiateStreaming` (#13925)

When an exception is thrown during the processing of streaming WebAssembly,
`op_wasm_streaming_abort` is called. This op calls into V8, which synchronously
rejects the promise and calls into the promise rejection handler, if applicable.
But calling an op borrows the isolate's `JsRuntimeState` for the duration of the
op, which means it is borrowed when V8 calls into `promise_reject_callback`,
which tries to borrow it again, panicking.

This change changes `op_wasm_streaming_abort` from an op to a binding
(`Deno.core.abortWasmStreaming`). Although that binding must borrow the
`JsRuntimeState` in order to access the `WasmStreamingResource` stored in the
`OpTable`, it also takes ownership of that `WasmStreamingResource` instance,
which means it can drop any borrows of the `JsRuntimeState` before calling into
V8.

6 files changed, 60 insertions, 26 deletions

diff --git a/cli/tests/integration/run_tests.rs b/cli/tests/integration/run_tests.rs
index 7c82acbf7..aec17faf0 100644
--- a/cli/tests/integration/run_tests.rs
+++ b/cli/tests/integration/run_tests.rs
@@ -2511,3 +2511,9 @@ itest!(config_not_auto_discovered_for_remote_script {
   output_str: Some("ok\n"),
   http_server: true,
 });
+
+itest!(wasm_streaming_panic_test {
+  args: "run wasm_streaming_panic_test.js",
+  output: "wasm_streaming_panic_test.js.out",
+  exit_code: 1,
+});
diff --git a/cli/tests/testdata/wasm_streaming_panic_test.js b/cli/tests/testdata/wasm_streaming_panic_test.js
new file mode 100644
index 000000000..ec017592f
--- /dev/null
+++ b/cli/tests/testdata/wasm_streaming_panic_test.js
@@ -0,0 +1,3 @@
+// https://github.com/denoland/deno/issues/13917
+
+WebAssembly.instantiateStreaming(Response.error());
diff --git a/cli/tests/testdata/wasm_streaming_panic_test.js.out b/cli/tests/testdata/wasm_streaming_panic_test.js.out
new file mode 100644
index 000000000..c21d709dd
--- /dev/null
+++ b/cli/tests/testdata/wasm_streaming_panic_test.js.out
@@ -0,0 +1,2 @@
+error: Uncaught (in promise) TypeError: Invalid WebAssembly content type.
+    at deno:ext/fetch/26_fetch.js:[WILDCARD]
diff --git a/core/bindings.rs b/core/bindings.rs
index 6e85a57d4..91087799a 100644
--- a/core/bindings.rs
+++ b/core/bindings.rs
@@ -7,10 +7,12 @@ use crate::modules::parse_import_assertions;
 use crate::modules::validate_import_assertions;
 use crate::modules::ImportAssertionsKind;
 use crate::modules::ModuleMap;
+use crate::ops_builtin::WasmStreamingResource;
 use crate::resolve_url_or_path;
 use crate::JsRuntime;
 use crate::OpState;
 use crate::PromiseId;
+use crate::ResourceId;
 use crate::ZeroCopyBuf;
 use anyhow::Error;
 use log::debug;
@@ -99,6 +101,9 @@ pub static EXTERNAL_REFERENCES: Lazy<v8::ExternalReferences> =
       v8::ExternalReference {
         function: set_wasm_streaming_callback.map_fn_to(),
       },
+      v8::ExternalReference {
+        function: abort_wasm_streaming.map_fn_to(),
+      },
     ])
   });
 
@@ -226,6 +231,7 @@ pub fn initialize_context<'s>(
     "setWasmStreamingCallback",
     set_wasm_streaming_callback,
   );
+  set_func(scope, core_val, "abortWasmStreaming", abort_wasm_streaming);
   // Direct bindings on `window`.
   set_func(scope, global, "queueMicrotask", queue_microtask);
 
@@ -762,8 +768,6 @@ fn set_wasm_streaming_callback(
   args: v8::FunctionCallbackArguments,
   _rv: v8::ReturnValue,
 ) {
-  use crate::ops_builtin::WasmStreamingResource;
-
   let cb = match arg0_to_cb(scope, args) {
     Ok(cb) => cb,
     Err(()) => return,
@@ -805,6 +809,48 @@ fn set_wasm_streaming_callback(
   });
 }
 
+fn abort_wasm_streaming(
+  scope: &mut v8::HandleScope,
+  args: v8::FunctionCallbackArguments,
+  _rv: v8::ReturnValue,
+) {
+  let rid: ResourceId = match serde_v8::from_v8(scope, args.get(0)) {
+    Ok(rid) => rid,
+    Err(_) => return throw_type_error(scope, "Invalid argument"),
+  };
+  let exception = args.get(1);
+
+  let wasm_streaming = {
+    let state_rc = JsRuntime::state(scope);
+    let state = state_rc.borrow();
+    let wsr = state
+      .op_state
+      .borrow_mut()
+      .resource_table
+      .take::<WasmStreamingResource>(rid);
+    match wsr {
+      Ok(wasm_streaming) => wasm_streaming,
+      Err(err) => {
+        let message = v8::String::new(scope, &err.to_string()).unwrap();
+        let v8_error = v8::Exception::error(scope, message);
+        scope.throw_exception(v8_error);
+        return;
+      }
+    }
+  };
+
+  // At this point there are no clones of Rc<WasmStreamingResource> on the
+  // resource table, and no one should own a reference because we're never
+  // cloning them. So we can be sure `wasm_streaming` is the only reference.
+  if let Ok(wsr) = std::rc::Rc::try_unwrap(wasm_streaming) {
+    // NOTE: v8::WasmStreaming::abort can't be called while `state` is borrowed;
+    // see https://github.com/denoland/deno/issues/13917
+    wsr.0.into_inner().abort(Some(exception));
+  } else {
+    panic!("Couldn't consume WasmStreamingResource.");
+  }
+}
+
 fn encode(
   scope: &mut v8::HandleScope,
   args: v8::FunctionCallbackArguments,
diff --git a/core/ops_builtin.rs b/core/ops_builtin.rs
index 23497ba36..4b566e916 100644
--- a/core/ops_builtin.rs
+++ b/core/ops_builtin.rs
@@ -26,7 +26,6 @@ pub(crate) fn init_builtins() -> Extension {
       op_print::decl(),
       op_resources::decl(),
       op_wasm_streaming_feed::decl(),
-      op_wasm_streaming_abort::decl(),
       op_wasm_streaming_set_url::decl(),
       op_void_sync::decl(),
       op_void_async::decl(),
@@ -142,28 +141,6 @@ pub fn op_wasm_streaming_feed(
   Ok(())
 }
 
-/// Abort a WasmStreamingResource.
-#[op]
-pub fn op_wasm_streaming_abort(
-  state: &mut OpState,
-  rid: ResourceId,
-  exception: serde_v8::Value,
-) -> Result<(), Error> {
-  let wasm_streaming =
-    state.resource_table.take::<WasmStreamingResource>(rid)?;
-
-  // At this point there are no clones of Rc<WasmStreamingResource> on the
-  // resource table, and no one should own a reference because we're never
-  // cloning them. So we can be sure `wasm_streaming` is the only reference.
-  if let Ok(wsr) = Rc::try_unwrap(wasm_streaming) {
-    wsr.0.into_inner().abort(Some(exception.v8_value));
-  } else {
-    panic!("Couldn't consume WasmStreamingResource.");
-  }
-
-  Ok(())
-}
-
 #[op]
 pub fn op_wasm_streaming_set_url(
   state: &mut OpState,
diff --git a/ext/fetch/26_fetch.js b/ext/fetch/26_fetch.js
index bbcbb44f0..711825985 100644
--- a/ext/fetch/26_fetch.js
+++ b/ext/fetch/26_fetch.js
@@ -556,7 +556,7 @@
         core.close(rid);
       } catch (err) {
         // 2.8 and 3
-        core.opSync("op_wasm_streaming_abort", rid, err);
+        core.abortWasmStreaming(rid, err);
       }
     })();
   }