feat(core): Reland support for async ops in realms (#17204)

Currently realms are supported on `deno_core`, but there was no support
for async ops anywhere other than the main realm. The main issue is that
the `js_recv_cb` callback, which resolves promises corresponding to
async ops, was only set for the main realm, so async ops in other realms
would never resolve. Furthermore, promise ID's are specific to each
realm, which meant that async ops from other realms would result in a
wrong promise from the main realm being resolved.

This change takes the `ContextState` struct added in #17050, and adds to
it a `js_recv_cb` callback for each realm. Combined with the fact that
that same PR also added a list of known realms to `JsRuntimeState`, and
that #17174 made `OpCtx` instances realm-specific and had them include
an index into that list of known realms, this makes it possible to know
the current realm in the `queue_async_op` and `queue_fast_async_op`
methods, and therefore to send the results of promises for each realm to
that realm, and prevent the ID's from getting mixed up.

Additionally, since promise ID's are no longer unique to the isolate,
having a single set of unrefed ops doesn't work. This change therefore
also moves `unrefed_ops` from `JsRuntimeState` to `ContextState`, and
adds the lengths of the unrefed op sets for all known realms to get the
total number of unrefed ops to compare in the event loop.

This PR is a reland of #14734 after it was reverted in #16366, except
that `ContextState` and `JsRuntimeState::known_realms` were previously
relanded in #17050. Another significant difference with the original PR
is passing around an index into `JsRuntimeState::known_realms` instead
of a `v8::Global<v8::Context>` to identify the realm, because async op
queuing in fast calls cannot call into V8, and therefore cannot have
access to V8 globals. This also simplified the implementation of
`resolve_async_ops`.

Co-authored-by: Luis Malheiro <luismalheiro@gmail.com>

core/ops.rs

@@ -91,6 +91,7 @@ where
   }
 }
 
+pub type RealmIdx = usize;
 pub type PromiseId = i32;
 pub type OpAsyncFuture = OpCall<(PromiseId, OpId, OpResult)>;
 pub type OpFn =
@@ -156,7 +157,7 @@ pub struct OpCtx {
   pub decl: Rc<OpDecl>,
   pub runtime_state: Weak<RefCell<JsRuntimeState>>,
   // Index of the current realm into `JsRuntimeState::known_realms`.
-  pub realm_idx: usize,
+  pub realm_idx: RealmIdx,
 }
 
 /// Maintains the resources and ops inside a JS runtime.

core/ops_builtin_v8.rs

@@ -76,14 +76,14 @@ fn to_v8_local_fn(
 
 #[op(v8)]
 fn op_ref_op(scope: &mut v8::HandleScope, promise_id: i32) {
-  let state_rc = JsRuntime::state(scope);
+  let context_state = JsRealm::state_from_scope(scope);
-  state_rc.borrow_mut().unrefed_ops.remove(&promise_id);
+  context_state.borrow_mut().unrefed_ops.remove(&promise_id);
 }
 
 #[op(v8)]
 fn op_unref_op(scope: &mut v8::HandleScope, promise_id: i32) {
-  let state_rc = JsRuntime::state(scope);
+  let context_state = JsRealm::state_from_scope(scope);
-  state_rc.borrow_mut().unrefed_ops.insert(promise_id);
+  context_state.borrow_mut().unrefed_ops.insert(promise_id);
 }
 
 #[op(v8)]

core/runtime.rs

@@ -48,7 +48,7 @@ use std::task::Context;
 use std::task::Poll;
 use v8::OwnedIsolate;
 
-type PendingOpFuture = OpCall<(PromiseId, OpId, OpResult)>;
+type PendingOpFuture = OpCall<(RealmIdx, PromiseId, OpId, OpResult)>;
 
 pub enum Snapshot {
   Static(&'static [u8]),
@@ -150,7 +150,11 @@ pub type CompiledWasmModuleStore = CrossIsolateStore<v8::CompiledWasmModule>;
 
 #[derive(Default)]
 pub(crate) struct ContextState {
+  js_recv_cb: Option<v8::Global<v8::Function>>,
   pub(crate) js_build_custom_error_cb: Option<v8::Global<v8::Function>>,
+  // TODO(andreubotella): Move the rest of Option<Global<Function>> fields from
+  // JsRuntimeState to this struct.
+  pub(crate) unrefed_ops: HashSet<i32>,
   // We don't explicitly re-read this prop but need the slice to live alongside
   // the context
   pub(crate) op_ctxs: Box<[OpCtx]>,
@@ -161,7 +165,6 @@ pub(crate) struct ContextState {
 pub struct JsRuntimeState {
   global_realm: Option<JsRealm>,
   known_realms: Vec<v8::Weak<v8::Context>>,
-  pub(crate) js_recv_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>>,
@@ -178,7 +181,6 @@ pub struct JsRuntimeState {
   pub(crate) source_map_getter: Option<Box<dyn SourceMapGetter>>,
   pub(crate) source_map_cache: SourceMapCache,
   pub(crate) pending_ops: FuturesUnordered<PendingOpFuture>,
-  pub(crate) unrefed_ops: HashSet<i32>,
   pub(crate) have_unpolled_ops: bool,
   pub(crate) op_state: Rc<RefCell<OpState>>,
   pub(crate) shared_array_buffer_store: Option<SharedArrayBufferStore>,
@@ -388,7 +390,6 @@ impl JsRuntime {
       pending_dyn_mod_evaluate: vec![],
       pending_mod_evaluate: None,
       dyn_module_evaluate_idle_counter: 0,
-      js_recv_cb: None,
       js_macrotask_cbs: vec![],
       js_nexttick_cbs: vec![],
       js_promise_reject_cb: None,
@@ -398,7 +399,6 @@ impl JsRuntime {
       source_map_getter: options.source_map_getter,
       source_map_cache: Default::default(),
       pending_ops: FuturesUnordered::new(),
-      unrefed_ops: HashSet::new(),
       shared_array_buffer_store: options.shared_array_buffer_store,
       compiled_wasm_module_store: options.compiled_wasm_module_store,
       op_state: op_state.clone(),
@@ -525,8 +525,8 @@ impl JsRuntime {
     global_context.open(&mut isolate).set_slot(
       &mut isolate,
       Rc::new(RefCell::new(ContextState {
-        js_build_custom_error_cb: None,
         op_ctxs,
+        ..Default::default()
       })),
     );
 
@@ -581,7 +581,8 @@ impl JsRuntime {
     let realm = js_runtime.global_realm();
     js_runtime.init_extension_js(&realm).unwrap();
     // Init callbacks (opresolve)
-    js_runtime.init_cbs();
+    let global_realm = js_runtime.global_realm();
+    js_runtime.init_cbs(&global_realm);
 
     js_runtime
   }
@@ -664,8 +665,8 @@ impl JsRuntime {
       context.set_slot(
         scope,
         Rc::new(RefCell::new(ContextState {
-          js_build_custom_error_cb: None,
           op_ctxs,
+          ..Default::default()
         })),
       );
 
@@ -679,7 +680,7 @@ impl JsRuntime {
     };
 
     self.init_extension_js(&realm)?;
-    self.init_realm_cbs(&realm);
+    self.init_cbs(&realm);
     Ok(realm)
   }
 
@@ -845,37 +846,25 @@ impl JsRuntime {
   }
 
   /// Grabs a reference to core.js' opresolve & syncOpsCache()
-  fn init_cbs(&mut self) {
+  fn init_cbs(&mut self, realm: &JsRealm) {
-    {
+    let (recv_cb, build_custom_error_cb) = {
-      let scope = &mut self.handle_scope();
+      let scope = &mut realm.handle_scope(self.v8_isolate());
       let recv_cb =
         Self::eval::<v8::Function>(scope, "Deno.core.opresolve").unwrap();
-      let recv_cb = v8::Global::new(scope, recv_cb);
-      // Put global handle in state
-      let state_rc = JsRuntime::state(scope);
-      let mut state = state_rc.borrow_mut();
-      state.js_recv_cb.replace(recv_cb);
-    }
-
-    // Also run init_realm_cbs for the main realm.
-    // TODO(@andreubotella): Merge this method back with `init_realm_cbs` when
-    // `js_recv_cb` is moved to ContextState.
-    let global_realm = self.global_realm();
-    self.init_realm_cbs(&global_realm);
-  }
-
-  fn init_realm_cbs(&mut self, realm: &JsRealm) {
-    let build_custom_error_cb = {
-      let scope = &mut realm.handle_scope(self.v8_isolate());
       let build_custom_error_cb =
         Self::eval::<v8::Function>(scope, "Deno.core.buildCustomError")
           .expect("Deno.core.buildCustomError is undefined in the realm");
-      v8::Global::new(scope, build_custom_error_cb)
+      (
+        v8::Global::new(scope, recv_cb),
+        v8::Global::new(scope, build_custom_error_cb),
+      )
     };
-    // Put global handle in the realm's ContextState
+
-    let state = realm.state(self.v8_isolate());
+    // Put global handles in the realm's ContextState
+    let state_rc = realm.state(self.v8_isolate());
+    let mut state = state_rc.borrow_mut();
+    state.js_recv_cb.replace(recv_cb);
     state
-      .borrow_mut()
       .js_build_custom_error_cb
       .replace(build_custom_error_cb);
   }
@@ -944,6 +933,7 @@ impl JsRuntime {
         if let Some(context) = weak_context.to_global(v8_isolate) {
           let realm = JsRealm::new(context.clone());
           let realm_state = realm.state(v8_isolate);
+          std::mem::take(&mut realm_state.borrow_mut().js_recv_cb);
           std::mem::take(
             &mut realm_state.borrow_mut().js_build_custom_error_cb,
           );
@@ -952,7 +942,6 @@ impl JsRuntime {
       }
 
       let mut state = self.state.borrow_mut();
-      std::mem::take(&mut state.js_recv_cb);
       std::mem::take(&mut state.js_promise_reject_cb);
       std::mem::take(&mut state.js_format_exception_cb);
       std::mem::take(&mut state.js_wasm_streaming_cb);
@@ -1387,8 +1376,16 @@ impl EventLoopPendingState {
     state: &mut JsRuntimeState,
     module_map: &ModuleMap,
   ) -> EventLoopPendingState {
+    let mut num_unrefed_ops = 0;
+    for weak_context in &state.known_realms {
+      if let Some(context) = weak_context.to_global(isolate) {
+        let realm = JsRealm(context);
+        num_unrefed_ops += realm.state(isolate).borrow().unrefed_ops.len();
+      }
+    }
+
     EventLoopPendingState {
-      has_pending_refed_ops: state.pending_ops.len() > state.unrefed_ops.len(),
+      has_pending_refed_ops: state.pending_ops.len() > num_unrefed_ops,
       has_pending_dyn_imports: module_map.has_pending_dynamic_imports(),
       has_pending_dyn_module_evaluation: !state
         .pending_dyn_mod_evaluate
@@ -2160,6 +2157,93 @@ impl JsRuntime {
 
   // Send finished responses to JS
   fn resolve_async_ops(&mut self, cx: &mut Context) -> Result<(), Error> {
+    // We have a specialized implementation of this method for the common case
+    // where there is only one realm.
+    let num_realms = self.state.borrow().known_realms.len();
+    if num_realms == 1 {
+      return self.resolve_single_realm_async_ops(cx);
+    }
+
+    // `responses_per_realm[idx]` is a vector containing the promise ID and
+    // response for all promises in realm `self.state.known_realms[idx]`.
+    let mut responses_per_realm: Vec<Vec<(PromiseId, OpResult)>> =
+      (0..num_realms).map(|_| vec![]).collect();
+
+    // Now handle actual ops.
+    {
+      let mut state = self.state.borrow_mut();
+      state.have_unpolled_ops = false;
+
+      while let Poll::Ready(Some(item)) = state.pending_ops.poll_next_unpin(cx)
+      {
+        let (realm_idx, promise_id, op_id, resp) = item;
+        state.op_state.borrow().tracker.track_async_completed(op_id);
+        responses_per_realm[realm_idx].push((promise_id, resp));
+      }
+    }
+
+    // Handle responses for each realm.
+    let isolate = self.v8_isolate.as_mut().unwrap();
+    for (realm_idx, responses) in responses_per_realm.into_iter().enumerate() {
+      if responses.is_empty() {
+        continue;
+      }
+
+      let realm = {
+        let context = self.state.borrow().known_realms[realm_idx]
+          .to_global(isolate)
+          .unwrap();
+        JsRealm(context)
+      };
+      let context_state_rc = realm.state(isolate);
+      let mut context_state = context_state_rc.borrow_mut();
+      let scope = &mut realm.handle_scope(isolate);
+
+      // We return async responses to JS in unbounded batches (may change),
+      // each batch is a flat vector of tuples:
+      // `[promise_id1, op_result1, promise_id2, op_result2, ...]`
+      // promise_id is a simple integer, op_result is an ops::OpResult
+      // which contains a value OR an error, encoded as a tuple.
+      // This batch is received in JS via the special `arguments` variable
+      // and then each tuple is used to resolve or reject promises
+      //
+      // This can handle 16 promises (32 / 2) futures in a single batch without heap
+      // allocations.
+      let mut args: SmallVec<[v8::Local<v8::Value>; 32]> =
+        SmallVec::with_capacity(responses.len() * 2);
+
+      for (promise_id, mut resp) in responses {
+        context_state.unrefed_ops.remove(&promise_id);
+        args.push(v8::Integer::new(scope, promise_id).into());
+        args.push(match resp.to_v8(scope) {
+          Ok(v) => v,
+          Err(e) => OpResult::Err(OpError::new(&|_| "TypeError", e.into()))
+            .to_v8(scope)
+            .unwrap(),
+        });
+      }
+
+      let js_recv_cb_handle = context_state.js_recv_cb.clone().unwrap();
+      let tc_scope = &mut v8::TryCatch::new(scope);
+      let js_recv_cb = js_recv_cb_handle.open(tc_scope);
+      let this = v8::undefined(tc_scope).into();
+      drop(context_state);
+      js_recv_cb.call(tc_scope, this, args.as_slice());
+
+      if let Some(exception) = tc_scope.exception() {
+        // TODO(@andreubotella): Returning here can cause async ops in other
+        // realms to never resolve.
+        return exception_to_err_result(tc_scope, exception, false);
+      }
+    }
+
+    Ok(())
+  }
+
+  fn resolve_single_realm_async_ops(
+    &mut self,
+    cx: &mut Context,
+  ) -> Result<(), Error> {
     let isolate = self.v8_isolate.as_mut().unwrap();
     let scope = &mut self
       .state
@@ -2186,10 +2270,17 @@ impl JsRuntime {
       let mut state = self.state.borrow_mut();
       state.have_unpolled_ops = false;
 
+      let realm_state_rc = state.global_realm.as_ref().unwrap().state(scope);
+      let mut realm_state = realm_state_rc.borrow_mut();
+
       while let Poll::Ready(Some(item)) = state.pending_ops.poll_next_unpin(cx)
       {
-        let (promise_id, op_id, mut resp) = item;
+        let (realm_idx, promise_id, op_id, mut resp) = item;
-        state.unrefed_ops.remove(&promise_id);
+        debug_assert_eq!(
+          state.known_realms[realm_idx],
+          state.global_realm.as_ref().unwrap().context()
+        );
+        realm_state.unrefed_ops.remove(&promise_id);
         state.op_state.borrow().tracker.track_async_completed(op_id);
         args.push(v8::Integer::new(scope, promise_id).into());
         args.push(match resp.to_v8(scope) {
@@ -2205,7 +2296,12 @@ impl JsRuntime {
       return Ok(());
     }
 
-    let js_recv_cb_handle = self.state.borrow().js_recv_cb.clone().unwrap();
+    let js_recv_cb_handle = {
+      let state = self.state.borrow_mut();
+      let realm_state_rc = state.global_realm.as_ref().unwrap().state(scope);
+      let handle = realm_state_rc.borrow().js_recv_cb.clone().unwrap();
+      handle
+    };
     let tc_scope = &mut v8::TryCatch::new(scope);
     let js_recv_cb = js_recv_cb_handle.open(tc_scope);
     let this = v8::undefined(tc_scope).into();
@@ -2412,7 +2508,7 @@ impl JsRealm {
 #[inline]
 pub fn queue_fast_async_op(
   ctx: &OpCtx,
-  op: impl Future<Output = (PromiseId, OpId, OpResult)> + 'static,
+  op: impl Future<Output = (RealmIdx, PromiseId, OpId, OpResult)> + 'static,
 ) {
   let runtime_state = match ctx.runtime_state.upgrade() {
     Some(rc_state) => rc_state,
@@ -2430,7 +2526,7 @@ pub fn queue_async_op(
   ctx: &OpCtx,
   scope: &mut v8::HandleScope,
   deferred: bool,
-  op: impl Future<Output = (PromiseId, OpId, OpResult)> + 'static,
+  op: impl Future<Output = (RealmIdx, PromiseId, OpId, OpResult)> + 'static,
 ) {
   let runtime_state = match ctx.runtime_state.upgrade() {
     Some(rc_state) => rc_state,
@@ -2457,18 +2553,20 @@ pub fn queue_async_op(
     // const p = setPromise();
     // op.op_async(promiseId, ...); // Calls `opresolve`
     // return p;
-    EagerPollResult::Ready((promise_id, op_id, mut resp)) if !deferred => {
+    EagerPollResult::Ready((_, promise_id, op_id, mut resp)) if !deferred => {
+      let context_state_rc = JsRealm::state_from_scope(scope);
+      let context_state = context_state_rc.borrow();
+
       let args = &[
         v8::Integer::new(scope, promise_id).into(),
         resp.to_v8(scope).unwrap(),
       ];
 
-      let js_recv_cb_handle =
-        runtime_state.borrow().js_recv_cb.clone().unwrap();
       ctx.state.borrow_mut().tracker.track_async_completed(op_id);
 
       let tc_scope = &mut v8::TryCatch::new(scope);
-      let js_recv_cb = js_recv_cb_handle.open(tc_scope);
+      let js_recv_cb =
+        context_state.js_recv_cb.as_ref().unwrap().open(tc_scope);
       let this = v8::undefined(tc_scope).into();
       js_recv_cb.call(tc_scope, this, args);
     }
@@ -2611,11 +2709,11 @@ pub mod tests {
       )
       .unwrap();
     {
+      let realm = runtime.global_realm();
       let isolate = runtime.v8_isolate();
       let state_rc = JsRuntime::state(isolate);
-      let state = state_rc.borrow();
+      assert_eq!(state_rc.borrow().pending_ops.len(), 2);
-      assert_eq!(state.pending_ops.len(), 2);
+      assert_eq!(realm.state(isolate).borrow().unrefed_ops.len(), 0);
-      assert_eq!(state.unrefed_ops.len(), 0);
     }
     runtime
       .execute_script(
@@ -2627,11 +2725,11 @@ pub mod tests {
       )
       .unwrap();
     {
+      let realm = runtime.global_realm();
       let isolate = runtime.v8_isolate();
       let state_rc = JsRuntime::state(isolate);
-      let state = state_rc.borrow();
+      assert_eq!(state_rc.borrow().pending_ops.len(), 2);
-      assert_eq!(state.pending_ops.len(), 2);
+      assert_eq!(realm.state(isolate).borrow().unrefed_ops.len(), 2);
-      assert_eq!(state.unrefed_ops.len(), 2);
     }
     runtime
       .execute_script(
@@ -2643,11 +2741,11 @@ pub mod tests {
       )
       .unwrap();
     {
+      let realm = runtime.global_realm();
       let isolate = runtime.v8_isolate();
       let state_rc = JsRuntime::state(isolate);
-      let state = state_rc.borrow();
+      assert_eq!(state_rc.borrow().pending_ops.len(), 2);
-      assert_eq!(state.pending_ops.len(), 2);
+      assert_eq!(realm.state(isolate).borrow().unrefed_ops.len(), 0);
-      assert_eq!(state.unrefed_ops.len(), 0);
     }
   }
 
@@ -4354,6 +4452,143 @@ Deno.core.ops.op_async_serialize_object_with_numbers_as_keys({
     }
   }
 
+  #[tokio::test]
+  async fn js_realm_async_ops() {
+    // Test that returning a ZeroCopyBuf and throwing an exception from a async
+    // op result in objects with prototypes from the right realm. Note that we
+    // don't test the result of returning structs, because they will be
+    // serialized to objects with null prototype.
+
+    #[op]
+    async fn op_test(fail: bool) -> Result<ZeroCopyBuf, Error> {
+      if !fail {
+        Ok(ZeroCopyBuf::empty())
+      } else {
+        Err(crate::error::type_error("Test"))
+      }
+    }
+
+    let mut runtime = JsRuntime::new(RuntimeOptions {
+      extensions: vec![Extension::builder("test_ext")
+        .ops(vec![op_test::decl()])
+        .build()],
+      get_error_class_fn: Some(&|error| {
+        crate::error::get_custom_error_class(error).unwrap()
+      }),
+      ..Default::default()
+    });
+
+    let global_realm = runtime.global_realm();
+    let new_realm = runtime.create_realm().unwrap();
+
+    let mut rets = vec![];
+
+    // Test in both realms
+    for realm in [global_realm, new_realm].into_iter() {
+      let ret = realm
+        .execute_script(
+          runtime.v8_isolate(),
+          "",
+          r#"
+            Deno.core.initializeAsyncOps();
+            (async function () {
+              const buf = await Deno.core.ops.op_test(false);
+              let err;
+              try {
+                await Deno.core.ops.op_test(true);
+              } catch(e) {
+                err = e;
+              }
+              return buf instanceof Uint8Array && buf.byteLength === 0 &&
+                      err instanceof TypeError && err.message === "Test" ;
+            })();
+          "#,
+        )
+        .unwrap();
+      rets.push((realm, ret));
+    }
+
+    runtime.run_event_loop(false).await.unwrap();
+
+    for ret in rets {
+      let scope = &mut ret.0.handle_scope(runtime.v8_isolate());
+      let value = v8::Local::new(scope, ret.1);
+      let promise = v8::Local::<v8::Promise>::try_from(value).unwrap();
+      let result = promise.result(scope);
+
+      assert!(result.is_boolean() && result.is_true());
+    }
+  }
+
+  #[tokio::test]
+  async fn js_realm_ref_unref_ops() {
+    run_in_task(|cx| {
+      // Never resolves.
+      #[op]
+      async fn op_pending() {
+        futures::future::pending().await
+      }
+
+      let mut runtime = JsRuntime::new(RuntimeOptions {
+        extensions: vec![Extension::builder("test_ext")
+          .ops(vec![op_pending::decl()])
+          .build()],
+        ..Default::default()
+      });
+      let main_realm = runtime.global_realm();
+      let other_realm = runtime.create_realm().unwrap();
+
+      main_realm
+        .execute_script(
+          runtime.v8_isolate(),
+          "",
+          r#"
+            Deno.core.initializeAsyncOps();
+            var promise = Deno.core.ops.op_pending();
+          "#,
+        )
+        .unwrap();
+      other_realm
+        .execute_script(
+          runtime.v8_isolate(),
+          "",
+          r#"
+            Deno.core.initializeAsyncOps();
+            var promise = Deno.core.ops.op_pending();
+          "#,
+        )
+        .unwrap();
+      assert!(matches!(runtime.poll_event_loop(cx, false), Poll::Pending));
+
+      main_realm
+        .execute_script(
+          runtime.v8_isolate(),
+          "",
+          r#"
+            let promiseIdSymbol = Symbol.for("Deno.core.internalPromiseId");
+            Deno.core.unrefOp(promise[promiseIdSymbol]);
+          "#,
+        )
+        .unwrap();
+      assert!(matches!(runtime.poll_event_loop(cx, false), Poll::Pending));
+
+      other_realm
+        .execute_script(
+          runtime.v8_isolate(),
+          "",
+          r#"
+            let promiseIdSymbol = Symbol.for("Deno.core.internalPromiseId");
+            Deno.core.unrefOp(promise[promiseIdSymbol]);
+          "#,
+        )
+        .unwrap();
+      assert!(matches!(
+        runtime.poll_event_loop(cx, false),
+        Poll::Ready(Ok(()))
+      ));
+    });
+  }
+
   #[test]
   fn test_array_by_copy() {
     // Verify that "array by copy" proposal is enabled (https://github.com/tc39/proposal-change-array-by-copy)

ops/fast_call.rs

@@ -265,10 +265,12 @@ pub(crate) fn generate(
 
     let queue_future = if optimizer.returns_result {
       q!({
+        let realm_idx = __ctx.realm_idx;
         let __get_class = __state.get_error_class_fn;
         let result = _ops::queue_fast_async_op(__ctx, async move {
           let result = result.await;
           (
+            realm_idx,
             __promise_id,
             __op_id,
             _ops::to_op_result(__get_class, result),
@@ -277,9 +279,15 @@ pub(crate) fn generate(
       })
     } else {
       q!({
+        let realm_idx = __ctx.realm_idx;
         let result = _ops::queue_fast_async_op(__ctx, async move {
           let result = result.await;
-          (__promise_id, __op_id, _ops::OpResult::Ok(result.into()))
+          (
+            realm_idx,
+            __promise_id,
+            __op_id,
+            _ops::OpResult::Ok(result.into()),
+          )
         });
       })
     };

ops/lib.rs

@@ -221,7 +221,7 @@ fn codegen_v8_async(
         quote! {
           let result = match result {
             Ok(fut) => fut.await,
-            Err(e) => return (promise_id, op_id, #core::_ops::to_op_result::<()>(get_class, Err(e))),
+            Err(e) => return (realm_idx, promise_id, op_id, #core::_ops::to_op_result::<()>(get_class, Err(e))),
           };
         }
       } else {
@@ -240,6 +240,7 @@ fn codegen_v8_async(
         as *const #core::_ops::OpCtx)
       };
       let op_id = ctx.id;
+      let realm_idx = ctx.realm_idx;
 
       let promise_id = args.get(0);
       let promise_id = #core::v8::Local::<#core::v8::Integer>::try_from(promise_id)
@@ -267,7 +268,7 @@ fn codegen_v8_async(
       #core::_ops::queue_async_op(ctx, scope, #deferred, async move {
         let result = #result_fut
         #result_wrapper
-        (promise_id, op_id, #core::_ops::to_op_result(get_class, result))
+        (realm_idx, promise_id, op_id, #core::_ops::to_op_result(get_class, result))
       });
     },
     argc,

ops/optimizer_tests/async_nop.out

@@ -43,6 +43,7 @@ impl op_void_async {
                 as *const deno_core::_ops::OpCtx)
         };
         let op_id = ctx.id;
+        let realm_idx = ctx.realm_idx;
         let promise_id = args.get(0);
         let promise_id = deno_core::v8::Local::<
             deno_core::v8::Integer,
@@ -71,7 +72,12 @@ impl op_void_async {
             async move {
                 let result = Self::call().await;
                 let result = Ok(result);
-                (promise_id, op_id, deno_core::_ops::to_op_result(get_class, result))
+                (
+                    realm_idx,
+                    promise_id,
+                    op_id,
+                    deno_core::_ops::to_op_result(get_class, result),
+                )
             },
         );
     }
@@ -111,11 +117,12 @@ fn op_void_async_fast_fn<'scope>(
     let __op_id = __ctx.id;
     let __state = ::std::cell::RefCell::borrow(&__ctx.state);
     __state.tracker.track_async(__op_id);
+    let realm_idx = __ctx.realm_idx;
     let result = _ops::queue_fast_async_op(
         __ctx,
         async move {
             let result = result.await;
-            (__promise_id, __op_id, _ops::OpResult::Ok(result.into()))
+            (realm_idx, __promise_id, __op_id, _ops::OpResult::Ok(result.into()))
         },
     );
     result

ops/optimizer_tests/async_result.out

@@ -47,6 +47,7 @@ impl op_read {
                 as *const deno_core::_ops::OpCtx)
         };
         let op_id = ctx.id;
+        let realm_idx = ctx.realm_idx;
         let promise_id = args.get(0);
         let promise_id = deno_core::v8::Local::<
             deno_core::v8::Integer,
@@ -130,7 +131,12 @@ impl op_read {
             false,
             async move {
                 let result = Self::call(ctx.state.clone(), arg_0, arg_1).await;
-                (promise_id, op_id, deno_core::_ops::to_op_result(get_class, result))
+                (
+                    realm_idx,
+                    promise_id,
+                    op_id,
+                    deno_core::_ops::to_op_result(get_class, result),
+                )
             },
         );
     }
@@ -179,12 +185,13 @@ fn op_read_fast_fn<'scope>(
     let __op_id = __ctx.id;
     let __state = ::std::cell::RefCell::borrow(&__ctx.state);
     __state.tracker.track_async(__op_id);
+    let realm_idx = __ctx.realm_idx;
     let __get_class = __state.get_error_class_fn;
     let result = _ops::queue_fast_async_op(
         __ctx,
         async move {
             let result = result.await;
-            (__promise_id, __op_id, _ops::to_op_result(__get_class, result))
+            (realm_idx, __promise_id, __op_id, _ops::to_op_result(__get_class, result))
         },
     );
 }

ops/optimizer_tests/issue16934.out

@@ -43,6 +43,7 @@ impl send_stdin {
                 as *const deno_core::_ops::OpCtx)
         };
         let op_id = ctx.id;
+        let realm_idx = ctx.realm_idx;
         let promise_id = args.get(0);
         let promise_id = deno_core::v8::Local::<
             deno_core::v8::Integer,
@@ -85,7 +86,12 @@ impl send_stdin {
                         arg_0,
                     )
                     .await;
-                (promise_id, op_id, deno_core::_ops::to_op_result(get_class, result))
+                (
+                    realm_idx,
+                    promise_id,
+                    op_id,
+                    deno_core::_ops::to_op_result(get_class, result),
+                )
             },
         );
     }

ops/optimizer_tests/issue16934_fast.out

@@ -41,6 +41,7 @@ impl send_stdin {
                 as *const deno_core::_ops::OpCtx)
         };
         let op_id = ctx.id;
+        let realm_idx = ctx.realm_idx;
         let promise_id = args.get(0);
         let promise_id = deno_core::v8::Local::<
             deno_core::v8::Integer,
@@ -83,7 +84,12 @@ impl send_stdin {
                         arg_0,
                     )
                     .await;
-                (promise_id, op_id, deno_core::_ops::to_op_result(get_class, result))
+                (
+                    realm_idx,
+                    promise_id,
+                    op_id,
+                    deno_core::_ops::to_op_result(get_class, result),
+                )
             },
         );
     }