Repository: denoland/deno_core
Pull request #350 (2023-11-24 20:21:04)

feat: V8TaskSpawner
Provides an `OpState`-stored (for now) `V8TaskSpawner` and `V8CrossThreadTaskSpawner` that allow for submission of v8 tasks to a queue that is run alongside ops.

```
let task_spawner: &V8TaskSpawn = op_state.borrow();
task_spawner.spawn(|scope| {
  /* use scope */
});
```

This allows you submit jobs into a queue that's run just before op dispatch (timing TBD and subject to change). The task is given a scope for the current context, and may do whatever it wishes.

This will likely turn into microtask submission API in the future.

mmastrac: > This is going to need a fair bit of tests to make sure that misbehaving tasks cannot break the event loop, I think.

I may need to be clearer about this as it's designed to replace the middleware system which is already a bit of a wild west. Effectively we need a way to run code _in_ the event loop right now that is free.
aapoalas: > > This is going to need a fair bit of tests to make sure that misbehaving tasks cannot break the event loop, I think.
> 
> I may need to be clearer about this as it's designed to replace the middleware system which is already a bit of a wild west. Effectively we need a way to run code _in_ the event loop right now that is free.

Yeah, I understand. Just, since we're reusing the same scope it means that user code can start affecting the event loop code's validity.

Patch:
diff --git a/Cargo.lock b/Cargo.lock
index 25601ae..eed9425 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -258,7 +258,7 @@ dependencies = [
 
 [[package]]
 name = "deno_core"
-version = "0.232.0"
+version = "0.233.0"
 dependencies = [
  "anyhow",
  "bencher",
@@ -278,6 +278,7 @@ dependencies = [
  "serde_v8",
  "smallvec",
  "sourcemap 7.0.1",
+ "static_assertions",
  "tokio",
  "url",
  "v8",
@@ -296,7 +297,7 @@ dependencies = [
 
 [[package]]
 name = "deno_ops"
-version = "0.108.0"
+version = "0.109.0"
 dependencies = [
  "pretty_assertions",
  "prettyplease",
@@ -1154,7 +1155,7 @@ dependencies = [
 
 [[package]]
 name = "serde_v8"
-version = "0.141.0"
+version = "0.142.0"
 dependencies = [
  "bencher",
  "bytes",
@@ -1888,9 +1889,9 @@ checksum = "5e395fcf16a7a3d8127ec99782007af141946b4795001f876d54fb0d55978560"
 
 [[package]]
 name = "v8"
-version = "0.81.0"
+version = "0.82.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b75f5f378b9b54aff3b10da8170d26af4cfd217f644cf671badcd13af5db4beb"
+checksum = "f53dfb242f4c0c39ed3fc7064378a342e57b5c9bd774636ad34ffe405b808121"
 dependencies = [
  "bitflags 1.3.2",
  "fslock",
diff --git a/Cargo.toml b/Cargo.toml
index dff5219..dc26ba6 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -17,11 +17,11 @@ repository = "https://github.com/denoland/deno_core"
 
 [workspace.dependencies]
 # Local dependencies
-deno_core = { version = "0.232.0", path = "./core" }
-deno_ops = { version = "0.108.0", path = "./ops" }
-serde_v8 = { version = "0.141.0", path = "./serde_v8" }
+deno_core = { version = "0.233.0", path = "./core" }
+deno_ops = { version = "0.109.0", path = "./ops" }
+serde_v8 = { version = "0.142.0", path = "./serde_v8" }
 
-v8 = { version = "0.81.0", default-features = false }
+v8 = { version = "0.82.0", default-features = false }
 deno_ast = { version = "0.31.2", features = ["transpiling"] }
 deno_unsync = "0.3.0"
 
@@ -44,6 +44,7 @@ serde_bytes = "0.11"
 serde_json = "1"
 smallvec = "1.8"
 sourcemap = "7"
+static_assertions = "1"
 strum = { version = "0.25.0", features = ["derive"] }
 strum_macros = "0.25.0"
 testing_macros = "0.2.11"
diff --git a/core/01_core.js b/core/01_core.js
index 2a174ec..0ba4536 100644
--- a/core/01_core.js
+++ b/core/01_core.js
@@ -750,6 +750,8 @@
     return { fileName, lineNumber, columnNumber };
   }
 
+  const hostObjectBrand = SymbolFor("Deno.core.hostObject");
+
   // Extra Deno.core.* exports
   const core = ObjectAssign(globalThis.Deno.core, {
     asyncStub,
@@ -794,7 +796,7 @@
       source,
       specifier,
     ) => ops.op_eval_context(source, specifier),
-    createHostObject: () => ops.op_create_host_object(),
+    hostObjectBrand,
     encode: (text) => ops.op_encode(text),
     decode: (buffer) => ops.op_decode(buffer),
     serialize: (
diff --git a/core/Cargo.toml b/core/Cargo.toml
index 369fbe5..21a0bb3 100644
--- a/core/Cargo.toml
+++ b/core/Cargo.toml
@@ -2,7 +2,7 @@
 
 [package]
 name = "deno_core"
-version = "0.232.0"
+version = "0.233.0"
 authors.workspace = true
 edition.workspace = true
 license.workspace = true
@@ -35,6 +35,7 @@ serde_json = { workspace = true, features = ["preserve_order"] }
 serde_v8.workspace = true
 smallvec.workspace = true
 sourcemap.workspace = true
+static_assertions.workspace = true
 tokio.workspace = true
 url.workspace = true
 v8.workspace = true
diff --git a/core/examples/fs_module_loader.rs b/core/examples/fs_module_loader.rs
index 737ff1d..ed58698 100644
--- a/core/examples/fs_module_loader.rs
+++ b/core/examples/fs_module_loader.rs
@@ -34,7 +34,7 @@ fn main() -> Result<(), Error> {
     let mod_id = js_runtime.load_main_module(&main_module, None).await?;
     let result = js_runtime.mod_evaluate(mod_id);
     js_runtime.run_event_loop(false).await?;
-    result.await?
+    result.await
   };
   runtime.block_on(future)
 }
diff --git a/core/examples/op2.rs b/core/examples/op2.rs
index 73bd01a..65ce99b 100644
--- a/core/examples/op2.rs
+++ b/core/examples/op2.rs
@@ -51,7 +51,7 @@ fn main() -> Result<(), Error> {
 
     let result = js_runtime.mod_evaluate(mod_id);
     js_runtime.run_event_loop(false).await?;
-    result.await?
+    result.await
   };
 
   tokio::runtime::Builder::new_current_thread()
diff --git a/core/examples/ts_module_loader.rs b/core/examples/ts_module_loader.rs
index 29a017b..a180406 100644
--- a/core/examples/ts_module_loader.rs
+++ b/core/examples/ts_module_loader.rs
@@ -157,7 +157,7 @@ fn main() -> Result<(), Error> {
     let mod_id = js_runtime.load_main_module(&main_module, None).await?;
     let result = js_runtime.mod_evaluate(mod_id);
     js_runtime.run_event_loop(false).await?;
-    result.await?
+    result.await
   };
 
   tokio::runtime::Builder::new_current_thread()
diff --git a/core/lib.rs b/core/lib.rs
index e3ea2b1..9dc0ca5 100644
--- a/core/lib.rs
+++ b/core/lib.rs
@@ -23,6 +23,7 @@ mod path;
 mod resources;
 mod runtime;
 mod source_map;
+mod tasks;
 
 // Re-exports
 pub use anyhow;
@@ -136,6 +137,8 @@ pub use crate::runtime::Snapshot;
 pub use crate::runtime::V8_WRAPPER_OBJECT_INDEX;
 pub use crate::runtime::V8_WRAPPER_TYPE_INDEX;
 pub use crate::source_map::SourceMapGetter;
+pub use crate::tasks::V8CrossThreadTaskSpawner;
+pub use crate::tasks::V8TaskSpawner;
 
 // Ensure we can use op2 in deno_core without any hackery.
 extern crate self as deno_core;
diff --git a/core/modules/map.rs b/core/modules/map.rs
index 8769cdb..59d7bcb 100644
--- a/core/modules/map.rs
+++ b/core/modules/map.rs
@@ -17,10 +17,13 @@ use crate::modules::ModuleRequest;
 use crate::modules::ModuleType;
 use crate::modules::NoopModuleLoader;
 use crate::modules::ResolutionKind;
+use crate::runtime::ContextState;
 use crate::runtime::JsRealm;
+use crate::runtime::JsRuntimeState;
 use crate::runtime::SnapshottedData;
 use crate::JsRuntime;
 use crate::ModuleSource;
+use anyhow::bail;
 use anyhow::Error;
 use futures::future::FutureExt;
 use futures::stream::FuturesUnordered;
@@ -36,12 +39,23 @@ use std::pin::Pin;
 use std::rc::Rc;
 use std::task::Context;
 use std::task::Poll;
+use tokio::sync::oneshot;
+
+use super::module_map_data::ModuleMapData;
+use super::AssertedModuleType;
 
 type PrepareLoadFuture =
   dyn Future<Output = (ModuleLoadId, Result<RecursiveModuleLoad, Error>)>;
 
-use super::module_map_data::ModuleMapData;
-use super::AssertedModuleType;
+use super::default_import_meta_resolve_cb;
+use super::ImportMetaResolveCallback;
+
+pub(crate) struct ModEvaluate {
+  promise: Option<v8::Global<v8::Promise>>,
+  pub(crate) has_evaluated: bool,
+  pub(crate) handled_promise_rejections: Vec<v8::Global<v8::Promise>>,
+  sender: oneshot::Sender<Result<(), Error>>,
+}
 
 pub const BOM_CHAR: &[u8] = &[0xef, 0xbb, 0xbf];
 
@@ -66,6 +80,7 @@ pub(crate) struct ModuleMap {
   // Handling of futures for loading module sources
   // TODO(mmastrac): we should not be swapping this loader out
   pub(crate) loader: RefCell<Rc<dyn ModuleLoader>>,
+  pub(crate) import_meta_resolve_cb: ImportMetaResolveCallback,
 
   dynamic_import_map:
     RefCell<HashMap<ModuleLoadId, v8::Global<v8::PromiseResolver>>>,
@@ -77,6 +92,7 @@ pub(crate) struct ModuleMap {
   pending_dynamic_imports_pending: Cell<bool>,
   pending_dyn_mod_evaluations: RefCell<Vec<DynImportModEvaluate>>,
   pending_dyn_mod_evaluations_pending: Cell<bool>,
+  pub(crate) pending_mod_evaluate: RefCell<Option<ModEvaluate>>,
   data: RefCell<ModuleMapData>,
 
   /// A counter used to delay our dynamic import deadlock detection by one spin
@@ -93,6 +109,21 @@ impl ModuleMap {
     id + 1
   }
 
+  pub(crate) fn notify_promise_rejection(
+    &self,
+    promise_global: v8::Global<v8::Promise>,
+  ) {
+    if let Some(pending_mod_evaluate) =
+      self.pending_mod_evaluate.borrow_mut().as_mut()
+    {
+      if !pending_mod_evaluate.has_evaluated {
+        pending_mod_evaluate
+          .handled_promise_rejections
+          .push(promise_global);
+      }
+    }
+  }
+
   #[cfg(debug_assertions)]
   pub(crate) fn assert_all_modules_evaluated(
     &self,
@@ -117,9 +148,13 @@ impl ModuleMap {
     }
   }
 
-  pub(crate) fn new(loader: Rc<dyn ModuleLoader>) -> Self {
+  pub(crate) fn new(
+    loader: Rc<dyn ModuleLoader>,
+    import_meta_resolve_cb: ImportMetaResolveCallback,
+  ) -> Self {
     Self {
       loader: loader.into(),
+      import_meta_resolve_cb,
       dyn_module_evaluate_idle_counter: Default::default(),
       dynamic_import_map: Default::default(),
       preparing_dynamic_imports: Default::default(),
@@ -128,16 +163,18 @@ impl ModuleMap {
       pending_dynamic_imports_pending: Default::default(),
       pending_dyn_mod_evaluations: Default::default(),
       pending_dyn_mod_evaluations_pending: Default::default(),
+      pending_mod_evaluate: Default::default(),
       data: Default::default(),
     }
   }
 
   pub(crate) fn new_from_snapshotted_data(
     loader: Rc<dyn ModuleLoader>,
+    import_meta_resolve_cb: ImportMetaResolveCallback,
     scope: &mut v8::HandleScope,
     data: SnapshottedData,
   ) -> Self {
-    let new = Self::new(loader);
+    let new = Self::new(loader, import_meta_resolve_cb);
     new
       .data
       .borrow_mut()
@@ -259,51 +296,36 @@ impl ModuleMap {
     Ok(module_id)
   }
 
-  pub(crate) fn new_json_module(
+  /// Creates a "synthetic module", that contains only a single, "default" export,
+  /// that returns the provided value.
+  pub fn new_synthetic_module(
     &self,
     scope: &mut v8::HandleScope,
     name: ModuleName,
-    source: ModuleCode,
+    module_type: ModuleType,
+    value: v8::Local<v8::Value>,
   ) -> Result<ModuleId, ModuleError> {
     let name_str = name.v8(scope);
-    let source_str = v8::String::new_from_utf8(
-      scope,
-      strip_bom(source.as_bytes()),
-      v8::NewStringType::Normal,
-    )
-    .unwrap();
-
-    let tc_scope = &mut v8::TryCatch::new(scope);
-
-    let parsed_json = match v8::json::parse(tc_scope, source_str) {
-      Some(parsed_json) => parsed_json,
-      None => {
-        assert!(tc_scope.has_caught());
-        let exception = tc_scope.exception().unwrap();
-        let exception = v8::Global::new(tc_scope, exception);
-        return Err(ModuleError::Exception(exception));
-      }
-    };
 
-    let export_names = [v8::String::new(tc_scope, "default").unwrap()];
+    let export_names = [v8::String::new(scope, "default").unwrap()];
     let module = v8::Module::create_synthetic_module(
-      tc_scope,
+      scope,
       name_str,
       &export_names,
-      json_module_evaluation_steps,
+      synthetic_module_evaluation_steps,
     );
 
-    let handle = v8::Global::<v8::Module>::new(tc_scope, module);
-    let value_handle = v8::Global::<v8::Value>::new(tc_scope, parsed_json);
+    let handle = v8::Global::<v8::Module>::new(scope, module);
+    let value_handle = v8::Global::<v8::Value>::new(scope, value);
     self
       .data
       .borrow_mut()
-      .json_value_store
+      .synthetic_module_value_store
       .insert(handle.clone(), value_handle);
 
     let id = self.data.borrow_mut().create_module_info(
       name,
-      ModuleType::Json,
+      module_type,
       handle,
       false,
       vec![],
@@ -417,6 +439,32 @@ impl ModuleMap {
     Ok(id)
   }
 
+  pub(crate) fn new_json_module(
+    &self,
+    scope: &mut v8::HandleScope,
+    name: ModuleName,
+    code: ModuleCode,
+  ) -> Result<ModuleId, ModuleError> {
+    let source_str = v8::String::new_from_utf8(
+      scope,
+      strip_bom(code.as_bytes()),
+      v8::NewStringType::Normal,
+    )
+    .unwrap();
+    let tc_scope = &mut v8::TryCatch::new(scope);
+
+    let parsed_json = match v8::json::parse(tc_scope, source_str) {
+      Some(parsed_json) => parsed_json,
+      None => {
+        assert!(tc_scope.has_caught());
+        let exception = tc_scope.exception().unwrap();
+        let exception = v8::Global::new(tc_scope, exception);
+        return Err(ModuleError::Exception(exception));
+      }
+    };
+    self.new_synthetic_module(tc_scope, name, ModuleType::Json, parsed_json)
+  }
+
   pub(crate) fn instantiate_module(
     &self,
     scope: &mut v8::HandleScope,
@@ -615,6 +663,195 @@ impl ModuleMap {
     self.pending_dyn_mod_evaluations_pending.get()
   }
 
+  /// Evaluates an already instantiated ES module.
+  ///
+  /// Returns a future that resolves when module promise resolves.
+  /// Implementors must manually call [`JsRuntime::run_event_loop`] to drive
+  /// module evaluation future.
+  ///
+  /// `Error` can usually be downcast to `JsError` and should be awaited and
+  /// checked after [`JsRuntime::run_event_loop`] completion.
+  ///
+  /// This function panics if module has not been instantiated.
+  pub fn mod_evaluate(
+    self: &Rc<Self>,
+    scope: &mut v8::HandleScope,
+    context_state: Rc<RefCell<ContextState>>,
+    runtime_state: Rc<JsRuntimeState>,
+    id: ModuleId,
+  ) -> impl Future<Output = Result<(), Error>> + Unpin {
+    let tc_scope = &mut v8::TryCatch::new(scope);
+
+    let module = self
+      .get_handle(id)
+      .map(|handle| v8::Local::new(tc_scope, handle))
+      .expect("ModuleInfo not found");
+    let mut status = module.get_status();
+    assert_eq!(
+      status,
+      v8::ModuleStatus::Instantiated,
+      "{} {} ({})",
+      if status == v8::ModuleStatus::Evaluated {
+        "Module already evaluated. Perhaps you've re-provided a module or extension that was already included in the snapshot?"
+      } else {
+        "Module not instantiated"
+      },
+      self.get_name_by_id(id).unwrap(),
+      id,
+    );
+
+    let (sender, receiver) = oneshot::channel();
+
+    // IMPORTANT: Top-level-await is enabled, which means that return value
+    // of module evaluation is a promise.
+    //
+    // Because that promise is created internally by V8, when error occurs during
+    // module evaluation the promise is rejected, and since the promise has no rejection
+    // handler it will result in call to `bindings::promise_reject_callback` adding
+    // the promise to pending promise rejection table - meaning JsRuntime will return
+    // error on next poll().
+    //
+    // This situation is not desirable as we want to manually return error at the
+    // end of this function to handle it further. It means we need to manually
+    // remove this promise from pending promise rejection table.
+    //
+    // For more details see:
+    // https://github.com/denoland/deno/issues/4908
+    // https://v8.dev/features/top-level-await#module-execution-order
+    {
+      let mut mod_evaluate = self.pending_mod_evaluate.borrow_mut();
+      assert!(
+        mod_evaluate.is_none(),
+        "There is already pending top level module evaluation"
+      );
+      *mod_evaluate = Some(ModEvaluate {
+        promise: None,
+        has_evaluated: false,
+        handled_promise_rejections: vec![],
+        sender,
+      });
+    }
+
+    let maybe_value = module.evaluate(tc_scope);
+
+    self
+      .pending_mod_evaluate
+      .borrow_mut()
+      .as_mut()
+      .unwrap()
+      .has_evaluated = true;
+
+    // Update status after evaluating.
+    status = module.get_status();
+
+    if runtime_state.has_dispatched_exception() {
+      // This will be overridden in `exception_to_err_result()`.
+      let exception = v8::undefined(tc_scope).into();
+      let pending_mod_evaluate =
+        self.pending_mod_evaluate.borrow_mut().take().unwrap();
+      pending_mod_evaluate
+        .sender
+        .send(exception_to_err_result(tc_scope, exception, false))
+        .expect("Failed to send module evaluation error.");
+    } else if let Some(value) = maybe_value {
+      assert!(
+        status == v8::ModuleStatus::Evaluated
+          || status == v8::ModuleStatus::Errored
+      );
+      let promise = v8::Local::<v8::Promise>::try_from(value)
+        .expect("Expected to get promise as module evaluation result");
+      let promise_global = v8::Global::new(tc_scope, promise);
+      {
+        let pending_mod_evaluate = self.pending_mod_evaluate.borrow_mut();
+        let pending_mod_evaluate = pending_mod_evaluate.as_ref().unwrap();
+        let pending_rejection_was_already_handled = pending_mod_evaluate
+          .handled_promise_rejections
+          .contains(&promise_global);
+        if !pending_rejection_was_already_handled {
+          context_state
+            .borrow_mut()
+            .pending_promise_rejections
+            .retain(|(key, _)| key != &promise_global);
+        }
+      }
+      let promise_global = v8::Global::new(tc_scope, promise);
+      self
+        .pending_mod_evaluate
+        .borrow_mut()
+        .as_mut()
+        .unwrap()
+        .promise = Some(promise_global);
+      tc_scope.perform_microtask_checkpoint();
+    } else if tc_scope.has_terminated() || tc_scope.is_execution_terminating() {
+      // Just drop the sender to send the generic closed error
+      _ = self.pending_mod_evaluate.borrow_mut().take();
+    } else {
+      assert!(status == v8::ModuleStatus::Errored);
+    }
+
+    receiver.map(|res| {
+      res.unwrap_or_else(|_| {
+        bail!("Cannot evaluate module, because JavaScript execution has been terminated")
+      }
+    )})
+  }
+
+  /// "deno_core" runs V8 with Top Level Await enabled. It means that each
+  /// module evaluation returns a promise from V8.
+  /// Feature docs: https://v8.dev/features/top-level-await
+  ///
+  /// This promise resolves after all dependent modules have also
+  /// resolved. Each dependent module may perform calls to "import()" and APIs
+  /// using async ops will add futures to the runtime's event loop.
+  /// It means that the promise returned from module evaluation will
+  /// resolve only after all futures in the event loop are done.
+  ///
+  /// Thus during turn of event loop we need to check if V8 has
+  /// resolved or rejected the promise. If the promise is still pending
+  /// then another turn of event loop must be performed.
+  pub(crate) fn evaluate_pending_module(&self, scope: &mut v8::HandleScope) {
+    let maybe_module_evaluation = self.pending_mod_evaluate.borrow_mut().take();
+    if maybe_module_evaluation.is_none() {
+      return;
+    }
+
+    let mut module_evaluation = maybe_module_evaluation.unwrap();
+
+    let promise_global = module_evaluation.promise.clone().unwrap();
+    let promise = promise_global.open(scope);
+    let promise_state = promise.state();
+    match promise_state {
+      v8::PromiseState::Pending => {
+        // NOTE: `poll_event_loop` will decide if
+        // runtime would be woken soon
+        *self.pending_mod_evaluate.borrow_mut() = Some(module_evaluation);
+      }
+      v8::PromiseState::Fulfilled => {
+        scope.perform_microtask_checkpoint();
+        // Receiver end might have been already dropped, ignore the result
+        let _ = module_evaluation.sender.send(Ok(()));
+        module_evaluation.handled_promise_rejections.clear();
+      }
+      v8::PromiseState::Rejected => {
+        let exception = promise.result(scope);
+        scope.perform_microtask_checkpoint();
+
+        // Receiver end might have been already dropped, ignore the result
+        if module_evaluation
+          .handled_promise_rejections
+          .contains(&promise_global)
+        {
+          let _ = module_evaluation.sender.send(Ok(()));
+          module_evaluation.handled_promise_rejections.clear();
+        } else {
+          let _ = module_evaluation
+            .sender
+            .send(exception_to_err_result(scope, exception, false));
+        }
+      }
+    }
+  }
+
   fn dynamic_import_module_evaluate(
     &self,
     scope: &mut v8::HandleScope,
@@ -1068,14 +1305,17 @@ impl ModuleMap {
 
 impl Default for ModuleMap {
   fn default() -> Self {
-    Self::new(Rc::new(NoopModuleLoader))
+    Self::new(
+      Rc::new(NoopModuleLoader),
+      Box::new(default_import_meta_resolve_cb),
+    )
   }
 }
 
 // Clippy thinks the return value doesn't need to be an Option, it's unaware
 // of the mapping that MapFnFrom<F> does for ResolveModuleCallback.
 #[allow(clippy::unnecessary_wraps)]
-fn json_module_evaluation_steps<'a>(
+fn synthetic_module_evaluation_steps<'a>(
   context: v8::Local<'a, v8::Context>,
   module: v8::Local<v8::Module>,
 ) -> Option<v8::Local<'a, v8::Value>> {
@@ -1088,7 +1328,7 @@ fn json_module_evaluation_steps<'a>(
   let value_handle = module_map
     .data
     .borrow_mut()
-    .json_value_store
+    .synthetic_module_value_store
     .remove(&handle)
     .unwrap();
   let value_local = v8::Local::new(tc_scope, value_handle);
diff --git a/core/modules/mod.rs b/core/modules/mod.rs
index 7c97639..82a0da2 100644
--- a/core/modules/mod.rs
+++ b/core/modules/mod.rs
@@ -1,6 +1,7 @@
 // Copyright 2018-2023 the Deno authors. All rights reserved. MIT license.
 use crate::fast_string::FastString;
 use crate::module_specifier::ModuleSpecifier;
+use anyhow::bail;
 use anyhow::Error;
 use serde::Deserialize;
 use serde::Serialize;
@@ -36,6 +37,23 @@ pub(crate) type ModuleLoadId = i32;
 pub type ModuleCode = FastString;
 pub type ModuleName = FastString;
 
+/// Callback to customize value of `import.meta.resolve("./foo.ts")`.
+pub type ImportMetaResolveCallback = Box<
+  dyn Fn(&dyn ModuleLoader, String, String) -> Result<ModuleSpecifier, Error>,
+>;
+
+pub(crate) fn default_import_meta_resolve_cb(
+  loader: &dyn ModuleLoader,
+  specifier: String,
+  referrer: String,
+) -> Result<ModuleSpecifier, Error> {
+  if specifier.starts_with("npm:") {
+    bail!("\"npm:\" specifiers are currently not supported in import.meta.resolve()");
+  }
+
+  loader.resolve(&specifier, &referrer, ResolutionKind::DynamicImport)
+}
+
 /// Callback to validate import attributes. If the validation fails and exception
 /// should be thrown using `scope.throw_exception()`.
 pub type ValidateImportAttributesCb =
diff --git a/core/modules/module_map_data.rs b/core/modules/module_map_data.rs
index cf1d1f4..96356a8 100644
--- a/core/modules/module_map_data.rs
+++ b/core/modules/module_map_data.rs
@@ -124,9 +124,9 @@ pub(crate) struct ModuleMapData {
   pub(crate) next_load_id: ModuleLoadId,
   /// If a main module has been loaded, points to it by index.
   pub(crate) main_module_id: Option<ModuleId>,
-  /// This store is used temporarily, to forward parsed JSON
-  /// value from `new_json_module` to `json_module_evaluation_steps`
-  pub(crate) json_value_store:
+  /// This store is used to temporarily store data that is used
+  /// to evaluate a "synthetic module".
+  pub(crate) synthetic_module_value_store:
     HashMap<v8::Global<v8::Module>, v8::Global<v8::Value>>,
 }
 
diff --git a/core/ops_builtin.rs b/core/ops_builtin.rs
index 90b4bbb..7a53f2c 100644
--- a/core/ops_builtin.rs
+++ b/core/ops_builtin.rs
@@ -53,7 +53,6 @@ deno_core::extension!(
     ops_builtin_v8::op_set_has_tick_scheduled,
     ops_builtin_v8::op_eval_context,
     ops_builtin_v8::op_queue_microtask,
-    ops_builtin_v8::op_create_host_object,
     ops_builtin_v8::op_encode,
     ops_builtin_v8::op_decode,
     ops_builtin_v8::op_serialize,
diff --git a/core/ops_builtin_v8.rs b/core/ops_builtin_v8.rs
index e4617dc..92c0aae 100644
--- a/core/ops_builtin_v8.rs
+++ b/core/ops_builtin_v8.rs
@@ -149,15 +149,6 @@ pub fn op_eval_context<'a>(
   }
 }
 
-#[op2]
-pub fn op_create_host_object<'a>(
-  scope: &mut v8::HandleScope<'a>,
-) -> v8::Local<'a, v8::Object> {
-  let template = v8::ObjectTemplate::new(scope);
-  template.set_internal_field_count(1);
-  template.new_instance(scope).unwrap()
-}
-
 #[op2]
 pub fn op_encode<'a>(
   scope: &mut v8::HandleScope<'a>,
@@ -208,6 +199,7 @@ struct SerializeDeserialize<'a> {
   host_objects: Option<v8::Local<'a, v8::Array>>,
   error_callback: Option<v8::Local<'a, v8::Function>>,
   for_storage: bool,
+  host_object_brand: Option<v8::Global<v8::Symbol>>,
 }
 
 impl<'a> v8::ValueSerializerImpl for SerializeDeserialize<'a> {
@@ -269,6 +261,23 @@ impl<'a> v8::ValueSerializerImpl for SerializeDeserialize<'a> {
     }
   }
 
+  fn has_custom_host_object(&mut self, _isolate: &mut v8::Isolate) -> bool {
+    true
+  }
+
+  fn is_host_object<'s>(
+    &mut self,
+    scope: &mut v8::HandleScope<'s>,
+    object: v8::Local<'s, v8::Object>,
+  ) -> Option<bool> {
+    if let Some(symbol) = &self.host_object_brand {
+      let key = v8::Local::new(scope, symbol);
+      object.has_own_property(scope, key.into())
+    } else {
+      Some(false)
+    }
+  }
+
   fn write_host_object<'s>(
     &mut self,
     scope: &mut v8::HandleScope<'s>,
@@ -393,10 +402,15 @@ pub fn op_serialize(
     None => None,
   };
 
+  let key = v8::String::new(scope, "Deno.core.hostObject").unwrap();
+  let symbol = v8::Symbol::for_key(scope, key);
+  let host_object_brand = Some(v8::Global::new(scope, symbol));
+
   let serialize_deserialize = Box::new(SerializeDeserialize {
     host_objects,
     error_callback,
     for_storage: options.for_storage,
+    host_object_brand,
   });
   let mut value_serializer =
     v8::ValueSerializer::new(scope, serialize_deserialize);
@@ -475,6 +489,7 @@ pub fn op_deserialize<'a>(
     host_objects,
     error_callback: None,
     for_storage: options.for_storage,
+    host_object_brand: None,
   });
   let mut value_deserializer =
     v8::ValueDeserializer::new(scope, serialize_deserialize, &zero_copy);
diff --git a/core/runtime/bindings.rs b/core/runtime/bindings.rs
index be97892..3ee0a70 100644
--- a/core/runtime/bindings.rs
+++ b/core/runtime/bindings.rs
@@ -13,7 +13,6 @@ use crate::modules::get_asserted_module_type_from_assertions;
 use crate::modules::parse_import_assertions;
 use crate::modules::ImportAssertionsKind;
 use crate::modules::ModuleMap;
-use crate::modules::ResolutionKind;
 use crate::ops::OpCtx;
 use crate::runtime::InitMode;
 use crate::runtime::JsRealm;
@@ -389,16 +388,13 @@ fn import_meta_resolve(
   let loader = module_map_rc.loader.clone();
   let specifier_str = specifier.to_rust_string_lossy(scope);
 
-  if specifier_str.starts_with("npm:") {
-    throw_type_error(scope, "\"npm:\" specifiers are currently not supported in import.meta.resolve()");
-    return;
-  }
+  let import_meta_resolve_result = {
+    let loader = loader.borrow();
+    let loader = loader.as_ref();
+    (module_map_rc.import_meta_resolve_cb)(loader, specifier_str, referrer)
+  };
 
-  match loader.borrow().resolve(
-    &specifier_str,
-    &referrer,
-    ResolutionKind::DynamicImport,
-  ) {
+  match import_meta_resolve_result {
     Ok(resolved) => {
       let resolved_val = serde_v8::to_v8(scope, resolved.as_str()).unwrap();
       rv.set(resolved_val);
@@ -508,15 +504,8 @@ pub extern "C" fn promise_reject_callback(message: v8::PromiseRejectMessage) {
       };
 
     if has_unhandled_rejection_handler {
-      let state_rc = JsRealm::state_from_scope(tc_scope);
-      let mut state = state_rc.borrow_mut();
-      if let Some(pending_mod_evaluate) = state.pending_mod_evaluate.as_mut() {
-        if !pending_mod_evaluate.has_evaluated {
-          pending_mod_evaluate
-            .handled_promise_rejections
-            .push(promise_global);
-        }
-      }
+      let modules = JsRealm::module_map_from(tc_scope);
+      modules.notify_promise_rejection(promise_global);
     }
   } else {
     let promise = message.get_promise();
diff --git a/core/runtime/jsrealm.rs b/core/runtime/jsrealm.rs
index e8dc591..2ff1825 100644
--- a/core/runtime/jsrealm.rs
+++ b/core/runtime/jsrealm.rs
@@ -1,7 +1,6 @@
 // Copyright 2018-2023 the Deno authors. All rights reserved. MIT license.
 use super::bindings;
 use crate::error::exception_to_err_result;
-use crate::error::generic_error;
 use crate::module_specifier::ModuleSpecifier;
 use crate::modules::ModuleCode;
 use crate::modules::ModuleError;
@@ -10,11 +9,12 @@ use crate::modules::ModuleMap;
 use crate::ops::OpCtx;
 use crate::ops::PendingOp;
 use crate::runtime::JsRuntimeState;
+use crate::tasks::V8TaskSpawnerFactory;
 use crate::JsRuntime;
 use anyhow::Error;
 use deno_unsync::JoinSet;
-use futures::channel::oneshot;
 use futures::stream::StreamExt;
+use futures::Future;
 use std::cell::RefCell;
 use std::collections::HashSet;
 use std::collections::VecDeque;
@@ -22,6 +22,7 @@ use std::hash::BuildHasherDefault;
 use std::hash::Hasher;
 use std::option::Option;
 use std::rc::Rc;
+use std::sync::Arc;
 use v8::Handle;
 use v8::HandleScope;
 use v8::Local;
@@ -45,15 +46,9 @@ impl Hasher for IdentityHasher {
   }
 }
 
-pub(crate) struct ModEvaluate {
-  promise: Option<v8::Global<v8::Promise>>,
-  pub(crate) has_evaluated: bool,
-  pub(crate) handled_promise_rejections: Vec<v8::Global<v8::Promise>>,
-  sender: oneshot::Sender<Result<(), Error>>,
-}
-
 #[derive(Default)]
 pub(crate) struct ContextState {
+  pub(crate) task_spawner_factory: Arc<V8TaskSpawnerFactory>,
   pub(crate) js_event_loop_tick_cb: Option<Rc<v8::Global<v8::Function>>>,
   pub(crate) js_build_custom_error_cb: Option<Rc<v8::Global<v8::Function>>>,
   pub(crate) js_promise_reject_cb: Option<Rc<v8::Global<v8::Function>>>,
@@ -61,7 +56,6 @@ pub(crate) struct ContextState {
   pub(crate) js_wasm_streaming_cb: Option<Rc<v8::Global<v8::Function>>>,
   pub(crate) pending_promise_rejections:
     VecDeque<(v8::Global<v8::Promise>, v8::Global<v8::Value>)>,
-  pub(crate) pending_mod_evaluate: Option<ModEvaluate>,
   pub(crate) unrefed_ops: HashSet<i32, BuildHasherDefault<IdentityHasher>>,
   pub(crate) pending_ops: JoinSet<PendingOp>,
   // We don't explicitly re-read this prop but need the slice to live alongside
@@ -362,10 +356,9 @@ impl JsRealm {
     self.0.module_map().instantiate_module(scope, id)
   }
 
-  // TODO(bartlomieju): make it return `ModuleEvaluationFuture`?
   /// Evaluates an already instantiated ES module.
   ///
-  /// Returns a receiver handle that resolves when module promise resolves.
+  /// Returns a future that resolves when module promise resolves.
   /// Implementors must manually call [`JsRuntime::run_event_loop`] to drive
   /// module evaluation future.
   ///
@@ -377,121 +370,13 @@ impl JsRealm {
     &self,
     scope: &mut v8::HandleScope,
     id: ModuleId,
-  ) -> oneshot::Receiver<Result<(), Error>> {
-    let state_rc = self.0.state();
-    let module_map_rc = self.0.module_map();
-    let tc_scope = &mut v8::TryCatch::new(scope);
-
-    let module = module_map_rc
-      .get_handle(id)
-      .map(|handle| v8::Local::new(tc_scope, handle))
-      .expect("ModuleInfo not found");
-    let mut status = module.get_status();
-    assert_eq!(
-      status,
-      v8::ModuleStatus::Instantiated,
-      "{} {} ({})",
-      if status == v8::ModuleStatus::Evaluated {
-        "Module already evaluated. Perhaps you've re-provided a module or extension that was already included in the snapshot?"
-      } else {
-        "Module not instantiated"
-      },
-      module_map_rc.get_name_by_id(id).unwrap(),
+  ) -> impl Future<Output = Result<(), Error>> + Unpin {
+    self.0.module_map.mod_evaluate(
+      scope,
+      self.0.context_state.clone(),
+      self.0.runtime_state.clone(),
       id,
-    );
-
-    let (sender, receiver) = oneshot::channel();
-
-    // IMPORTANT: Top-level-await is enabled, which means that return value
-    // of module evaluation is a promise.
-    //
-    // Because that promise is created internally by V8, when error occurs during
-    // module evaluation the promise is rejected, and since the promise has no rejection
-    // handler it will result in call to `bindings::promise_reject_callback` adding
-    // the promise to pending promise rejection table - meaning JsRuntime will return
-    // error on next poll().
-    //
-    // This situation is not desirable as we want to manually return error at the
-    // end of this function to handle it further. It means we need to manually
-    // remove this promise from pending promise rejection table.
-    //
-    // For more details see:
-    // https://github.com/denoland/deno/issues/4908
-    // https://v8.dev/features/top-level-await#module-execution-order
-    {
-      let mut state = state_rc.borrow_mut();
-      assert!(
-        state.pending_mod_evaluate.is_none(),
-        "There is already pending top level module evaluation"
-      );
-      state.pending_mod_evaluate = Some(ModEvaluate {
-        promise: None,
-        has_evaluated: false,
-        handled_promise_rejections: vec![],
-        sender,
-      });
-    }
-
-    let maybe_value = module.evaluate(tc_scope);
-    {
-      let mut state = state_rc.borrow_mut();
-      let pending_mod_evaluate = state.pending_mod_evaluate.as_mut().unwrap();
-      pending_mod_evaluate.has_evaluated = true;
-    }
-
-    // Update status after evaluating.
-    status = module.get_status();
-
-    let has_dispatched_exception =
-      self.0.runtime_state.has_dispatched_exception();
-    if has_dispatched_exception {
-      // This will be overridden in `exception_to_err_result()`.
-      let exception = v8::undefined(tc_scope).into();
-      let pending_mod_evaluate = {
-        let mut state = state_rc.borrow_mut();
-        state.pending_mod_evaluate.take().unwrap()
-      };
-      pending_mod_evaluate
-        .sender
-        .send(exception_to_err_result(tc_scope, exception, false))
-        .expect("Failed to send module evaluation error.");
-    } else if let Some(value) = maybe_value {
-      assert!(
-        status == v8::ModuleStatus::Evaluated
-          || status == v8::ModuleStatus::Errored
-      );
-      let promise = v8::Local::<v8::Promise>::try_from(value)
-        .expect("Expected to get promise as module evaluation result");
-      let promise_global = v8::Global::new(tc_scope, promise);
-      let mut state = state_rc.borrow_mut();
-      {
-        let pending_mod_evaluate = state.pending_mod_evaluate.as_ref().unwrap();
-        let pending_rejection_was_already_handled = pending_mod_evaluate
-          .handled_promise_rejections
-          .contains(&promise_global);
-        if !pending_rejection_was_already_handled {
-          state
-            .pending_promise_rejections
-            .retain(|(key, _)| key != &promise_global);
-        }
-      }
-      let promise_global = v8::Global::new(tc_scope, promise);
-      state.pending_mod_evaluate.as_mut().unwrap().promise =
-        Some(promise_global);
-      tc_scope.perform_microtask_checkpoint();
-    } else if tc_scope.has_terminated() || tc_scope.is_execution_terminating() {
-      let pending_mod_evaluate = {
-        let mut state = state_rc.borrow_mut();
-        state.pending_mod_evaluate.take().unwrap()
-      };
-      pending_mod_evaluate.sender.send(Err(
-        generic_error("Cannot evaluate module, because JavaScript execution has been terminated.")
-      )).expect("Failed to send module evaluation error.");
-    } else {
-      assert!(status == v8::ModuleStatus::Errored);
-    }
-
-    receiver
+    )
   }
 
   pub(crate) fn modules_idle(&self) -> bool {
@@ -503,71 +388,6 @@ impl JsRealm {
     count.set(count.get() + 1)
   }
 
-  /// "deno_core" runs V8 with Top Level Await enabled. It means that each
-  /// module evaluation returns a promise from V8.
-  /// Feature docs: https://v8.dev/features/top-level-await
-  ///
-  /// This promise resolves after all dependent modules have also
-  /// resolved. Each dependent module may perform calls to "import()" and APIs
-  /// using async ops will add futures to the runtime's event loop.
-  /// It means that the promise returned from module evaluation will
-  /// resolve only after all futures in the event loop are done.
-  ///
-  /// Thus during turn of event loop we need to check if V8 has
-  /// resolved or rejected the promise. If the promise is still pending
-  /// then another turn of event loop must be performed.
-  pub(in crate::runtime) fn evaluate_pending_module(
-    &self,
-    scope: &mut v8::HandleScope,
-  ) {
-    let maybe_module_evaluation = self
-      .0
-      .context_state
-      .borrow_mut()
-      .pending_mod_evaluate
-      .take();
-    if maybe_module_evaluation.is_none() {
-      return;
-    }
-
-    let mut module_evaluation = maybe_module_evaluation.unwrap();
-    let state_rc = self.0.state();
-
-    let promise_global = module_evaluation.promise.clone().unwrap();
-    let promise = promise_global.open(scope);
-    let promise_state = promise.state();
-    match promise_state {
-      v8::PromiseState::Pending => {
-        // NOTE: `poll_event_loop` will decide if
-        // runtime would be woken soon
-        state_rc.borrow_mut().pending_mod_evaluate = Some(module_evaluation);
-      }
-      v8::PromiseState::Fulfilled => {
-        scope.perform_microtask_checkpoint();
-        // Receiver end might have been already dropped, ignore the result
-        let _ = module_evaluation.sender.send(Ok(()));
-        module_evaluation.handled_promise_rejections.clear();
-      }
-      v8::PromiseState::Rejected => {
-        let exception = promise.result(scope);
-        scope.perform_microtask_checkpoint();
-
-        // Receiver end might have been already dropped, ignore the result
-        if module_evaluation
-          .handled_promise_rejections
-          .contains(&promise_global)
-        {
-          let _ = module_evaluation.sender.send(Ok(()));
-          module_evaluation.handled_promise_rejections.clear();
-        } else {
-          let _ = module_evaluation
-            .sender
-            .send(exception_to_err_result(scope, exception, false));
-        }
-      }
-    }
-  }
-
   /// Asynchronously load specified module and all of its dependencies.
   ///
   /// The module will be marked as "main", and because of that
diff --git a/core/runtime/jsruntime.rs b/core/runtime/jsruntime.rs
index b6f5b24..df04605 100644
--- a/core/runtime/jsruntime.rs
+++ b/core/runtime/jsruntime.rs
@@ -14,8 +14,10 @@ use crate::extensions::OpDecl;
 use crate::include_js_files;
 use crate::inspector::JsRuntimeInspector;
 use crate::module_specifier::ModuleSpecifier;
+use crate::modules::default_import_meta_resolve_cb;
 use crate::modules::AssertedModuleType;
 use crate::modules::ExtModuleLoader;
+use crate::modules::ImportMetaResolveCallback;
 use crate::modules::ModuleCode;
 use crate::modules::ModuleId;
 use crate::modules::ModuleLoader;
@@ -40,10 +42,11 @@ use crate::V8_WRAPPER_OBJECT_INDEX;
 use crate::V8_WRAPPER_TYPE_INDEX;
 use anyhow::Context as AnyhowContext;
 use anyhow::Error;
-use futures::channel::oneshot;
 use futures::future::poll_fn;
+use futures::task::noop_waker_ref;
 use futures::task::AtomicWaker;
 use futures::Future;
+use futures::FutureExt;
 use smallvec::SmallVec;
 use std::any::Any;
 use std::cell::Cell;
@@ -444,9 +447,15 @@ pub struct RuntimeOptions {
   pub feature_checker: Option<Arc<FeatureChecker>>,
 
   /// A callback that can be used to validate import attributes received at
-  /// the import site. If not callback is provided, a default one is used. The
+  /// the import site. If no callback is provided, a default one is used. The
   /// default callback only allows `"type"` attribute, with a value of `"json"`.
   pub validate_import_attributes_cb: Option<ValidateImportAttributesCb>,
+
+  /// A callback that can be used to customize behavior of
+  /// `import.meta.resolve()` API. If no callback is provided, a default one
+  /// is used. The default callback returns value of
+  /// `RuntimeOptions::module_loader::resolve()` call.
+  pub import_meta_resolve_callback: Option<ImportMetaResolveCallback>,
 }
 
 impl RuntimeOptions {
@@ -623,6 +632,21 @@ impl JsRuntime {
 
     let weak = Rc::downgrade(&state_rc);
     let context_state = Rc::new(RefCell::new(ContextState::default()));
+
+    // Add the task spawners to the OpState
+    let spawner = context_state
+      .borrow()
+      .task_spawner_factory
+      .clone()
+      .new_same_thread_spawner();
+    op_state.borrow_mut().put(spawner);
+    let spawner = context_state
+      .borrow()
+      .task_spawner_factory
+      .clone()
+      .new_cross_thread_spawner();
+    op_state.borrow_mut().put(spawner);
+
     let count = ops.len();
     let mut op_ctxs = ops
       .into_iter()
@@ -743,15 +767,19 @@ impl JsRuntime {
     let loader = options
       .module_loader
       .unwrap_or_else(|| Rc::new(NoopModuleLoader));
+    let import_meta_resolve_cb = options
+      .import_meta_resolve_callback
+      .unwrap_or_else(|| Box::new(default_import_meta_resolve_cb));
 
     let module_map = if let Some(snapshotted_data) = snapshotted_data {
       Rc::new(ModuleMap::new_from_snapshotted_data(
         loader,
+        import_meta_resolve_cb,
         scope,
         snapshotted_data,
       ))
     } else {
-      Rc::new(ModuleMap::new(loader))
+      Rc::new(ModuleMap::new(loader, import_meta_resolve_cb))
     };
 
     context.set_slot(scope, module_map.clone());
@@ -933,19 +961,19 @@ impl JsRuntime {
           let isolate = self.v8_isolate();
           let scope = &mut realm.handle_scope(isolate);
           let receiver = realm.mod_evaluate(scope, mod_id);
-          realm.evaluate_pending_module(scope);
+          module_map.evaluate_pending_module(scope);
           receiver
         };
 
         // After evaluate_pending_module, if the module isn't fully evaluated
         // and the resolver solved, it means the module or one of its imports
         // uses TLA.
-        match receiver.try_recv()? {
-          Some(result) => {
+        match receiver.poll_unpin(&mut Context::from_waker(noop_waker_ref())) {
+          Poll::Ready(result) => {
             result
               .with_context(|| format!("Couldn't execute '{specifier}'"))?;
           }
-          None => {
+          Poll::Pending => {
             // Find the TLA location to display it on the panic.
             let location = {
               let scope = &mut realm.handle_scope(self.v8_isolate());
@@ -1554,7 +1582,7 @@ impl JsRuntime {
     }
 
     // Top level module
-    realm.evaluate_pending_module(scope);
+    modules.evaluate_pending_module(scope);
 
     // Get the pending state from the main realm, or all realms
     let pending_state = {
@@ -1772,12 +1800,15 @@ impl EventLoopPendingState {
   ) -> Self {
     let num_unrefed_ops = state.unrefed_ops.len();
     let num_pending_ops = state.pending_ops.len();
+    let has_pending_tasks = state.task_spawner_factory.has_pending_tasks();
     let has_pending_dyn_imports = modules.has_pending_dynamic_imports();
     let has_pending_dyn_module_evaluation =
       modules.has_pending_dyn_module_evaluation();
-    let has_pending_module_evaluation = state.pending_mod_evaluate.is_some();
+    let has_pending_module_evaluation =
+      modules.pending_mod_evaluate.borrow().is_some();
     EventLoopPendingState {
-      has_pending_refed_ops: num_pending_ops > num_unrefed_ops,
+      has_pending_refed_ops: has_pending_tasks
+        || num_pending_ops > num_unrefed_ops,
       has_pending_dyn_imports,
       has_pending_dyn_module_evaluation,
       has_pending_module_evaluation,
@@ -1895,10 +1926,9 @@ impl JsRuntime {
     realm.instantiate_module(scope, id)
   }
 
-  // TODO(bartlomieju): make it return `ModuleEvaluationFuture`?
   /// Evaluates an already instantiated ES module.
   ///
-  /// Returns a receiver handle that resolves when module promise resolves.
+  /// Returns a future that resolves when module promise resolves.
   /// Implementors must manually call [`JsRuntime::run_event_loop`] to drive
   /// module evaluation future.
   ///
@@ -1909,7 +1939,7 @@ impl JsRuntime {
   pub fn mod_evaluate(
     &mut self,
     id: ModuleId,
-  ) -> oneshot::Receiver<Result<(), Error>> {
+  ) -> impl Future<Output = Result<(), Error>> {
     let isolate = &mut self.inner.v8_isolate;
     let realm = &self.inner.main_realm;
     let scope = &mut realm.handle_scope(isolate);
@@ -1961,9 +1991,21 @@ impl JsRuntime {
     scope: &mut v8::HandleScope,
     context_state: &RefCell<ContextState>,
   ) -> Result<bool, Error> {
-    let mut context_state = context_state.borrow_mut();
     let mut dispatched_ops = false;
 
+    // Poll any pending task spawner tasks. Note that we need to poll separately because otherwise
+    // Rust will extend the lifetime of the borrow longer than we expect.
+    let tasks = context_state.borrow().task_spawner_factory.poll_inner(cx);
+    if let Poll::Ready(tasks) = tasks {
+      // TODO(mmastrac): we are using this flag
+      dispatched_ops = true;
+      for task in tasks {
+        task(scope);
+      }
+    }
+
+    let mut context_state = context_state.borrow_mut();
+
     // 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, ...]`
diff --git a/core/tasks.rs b/core/tasks.rs
new file mode 100644
index 0000000..e7a7fad
--- /dev/null
+++ b/core/tasks.rs
@@ -0,0 +1,286 @@
+// Copyright 2018-2023 the Deno authors. All rights reserved. MIT license.
+
+use futures::task::AtomicWaker;
+use std::marker::PhantomData;
+use std::ops::DerefMut;
+use std::sync::atomic::AtomicBool;
+use std::sync::atomic::Ordering;
+use std::sync::Arc;
+use std::sync::Mutex;
+use std::task::Context;
+use std::task::Poll;
+
+type UnsendTask = Box<dyn FnOnce(&mut v8::HandleScope) + 'static>;
+type SendTask = Box<dyn FnOnce(&mut v8::HandleScope) + Send + 'static>;
+
+static_assertions::assert_not_impl_any!(V8TaskSpawnerFactory: Send);
+static_assertions::assert_not_impl_any!(V8TaskSpawner: Send);
+static_assertions::assert_impl_all!(V8CrossThreadTaskSpawner: Send);
+
+/// The [`V8TaskSpawnerFactory`] must be created on the same thread as the thread that runs tasks.
+///
+/// This factory is not [`Send`] because it may contain `!Send` tasks submitted by a
+/// [`V8CrossThreadTaskSpawner`]. It is only safe to send this object to another thread if you plan on
+/// submitting [`Send`] tasks to it, which is what [`V8CrossThreadTaskSpawner`] does.
+#[derive(Default)]
+pub(crate) struct V8TaskSpawnerFactory {
+  // TODO(mmastrac): ideally we wouldn't box if we could use arena allocation and a max submission size
+  // TODO(mmastrac): we may want to split the Send and !Send tasks
+  /// The set of tasks, non-empty if `has_tasks` is set.
+  tasks: Mutex<Vec<SendTask>>,
+  /// A flag we can poll without any locks.
+  has_tasks: AtomicBool,
+  /// The polled waker, woken on task submission.
+  waker: AtomicWaker,
+  /// Mark as `!Send`. See note above.
+  _unsend_marker: PhantomData<*const ()>,
+}
+
+impl V8TaskSpawnerFactory {
+  pub fn new_same_thread_spawner(self: Arc<Self>) -> V8TaskSpawner {
+    V8TaskSpawner {
+      tasks: self,
+      _unsend_marker: PhantomData,
+    }
+  }
+
+  pub fn new_cross_thread_spawner(self: Arc<Self>) -> V8CrossThreadTaskSpawner {
+    V8CrossThreadTaskSpawner { tasks: self }
+  }
+
+  /// `false` guarantees that there are no queued tasks, while `true` means that it is likely (but not guaranteed)
+  /// that tasks exist.
+  ///
+  /// Calls should prefer using the waker, but this is left while we rework the event loop.
+  pub fn has_pending_tasks(&self) -> bool {
+    // Ensure that all reads after this point happen-after we load from the atomic
+    self.has_tasks.load(Ordering::Acquire)
+  }
+
+  /// Poll this set of tasks, returning a non-empty set of tasks if there have
+  /// been any queued, or registering the waker if not.
+  pub fn poll_inner(&self, cx: &mut Context) -> Poll<Vec<UnsendTask>> {
+    // Check the flag first -- if it's false we definitely have no tasks. AcqRel semantics ensure
+    // that this read happens-before the vector read below, and that any writes happen-after the success
+    // case.
+    if self
+      .has_tasks
+      .compare_exchange(true, false, Ordering::AcqRel, Ordering::Acquire)
+      .is_err()
+    {
+      self.waker.register(cx.waker());
+      return Poll::Pending;
+    }
+
+    let mut lock = self.tasks.lock().unwrap();
+    let tasks = std::mem::take(lock.deref_mut());
+    if tasks.is_empty() {
+      // Unlikely race lost -- the task submission to the queue and flag are not atomic, so it's
+      // possible we ended up with an extra poll here. This only shows up under Miri, but as it is
+      // possible we do need to handle it.
+      self.waker.register(cx.waker());
+      return Poll::Pending;
+    }
+
+    // SAFETY: we are removing the Send trait as we return the tasks here to prevent
+    // these tasks from accidentally leaking to another thread.
+    let tasks = unsafe { std::mem::transmute(tasks) };
+    Poll::Ready(tasks)
+  }
+
+  fn spawn(&self, task: SendTask) {
+    self.tasks.lock().unwrap().push(task);
+    // TODO(mmastrac): can we skip the mutex here?
+    // Release ordering means that the writes in the above lock happen-before the atomic store
+    self.has_tasks.store(true, Ordering::Release);
+    self.waker.wake();
+  }
+}
+
+/// Allows for submission of v8 tasks on the same thread.
+#[derive(Clone)]
+pub struct V8TaskSpawner {
+  // TODO(mmastrac): can we split the waker into a send and !send one?
+  tasks: Arc<V8TaskSpawnerFactory>,
+  _unsend_marker: PhantomData<*const ()>,
+}
+
+impl V8TaskSpawner {
+  /// Spawn a task that runs within the [`crate::JsRuntime`] event loop from the same thread
+  /// that the runtime is running on. This function is re-entrant-safe and may be called from
+  /// ops, from outside of a [`v8::HandleScope`] in a plain `async`` task, or even from within
+  /// another, previously-spawned task.
+  ///
+  /// The task is handed off to be run the next time the event loop is polled, and there are
+  /// no guarantees as to when this may happen.
+  ///
+  /// # Safety
+  ///
+  /// The task shares the same [`v8::HandleScope`] as the core event loop, which means that it
+  /// must maintain the scope in a valid state to avoid corrupting or destroying the runtime.
+  ///
+  /// For example, if the code called by this task can raise an exception, the task must ensure
+  /// that it calls that code within a new [`v8::TryCatch`] to avoid the exception leaking to the
+  /// event loop's [`v8::HandleScope`].
+  pub fn spawn<F>(&self, f: F)
+  where
+    F: FnOnce(&mut v8::HandleScope) + 'static,
+  {
+    let task: Box<dyn FnOnce(&mut v8::HandleScope<'_>)> = Box::new(f);
+    // SAFETY: we are transmuting Send into a !Send handle but we guarantee this object will never
+    // leave the current thread because `V8TaskSpawner` is !Send.
+    let task: Box<dyn FnOnce(&mut v8::HandleScope<'_>) + Send> =
+      unsafe { std::mem::transmute(task) };
+    self.tasks.spawn(task)
+  }
+}
+
+/// Allows for submission of v8 tasks on any thread.
+#[derive(Clone)]
+pub struct V8CrossThreadTaskSpawner {
+  tasks: Arc<V8TaskSpawnerFactory>,
+}
+
+// SAFETY: the underlying V8TaskSpawnerFactory is not Send, but we always submit Send tasks
+// to it from this spawner.
+unsafe impl Send for V8CrossThreadTaskSpawner {}
+
+impl V8CrossThreadTaskSpawner {
+  /// Spawn a task that runs within the [`crate::JsRuntime`] event loop, potentially (but not
+  /// required to be) from a different thread than the runtime is running on.
+  ///
+  /// The task is handed off to be run the next time the event loop is polled, and there are
+  /// no guarantees as to when this may happen.
+  ///
+  /// # Safety
+  ///
+  /// The task shares the same [`v8::HandleScope`] as the core event loop, which means that it
+  /// must maintain the scope in a valid state to avoid corrupting or destroying the runtime.
+  ///
+  /// For example, if the code called by this task can raise an exception, the task must ensure
+  /// that it calls that code within a new [`v8::TryCatch`] to avoid the exception leaking to the
+  /// event loop's [`v8::HandleScope`].
+  pub fn spawn<F>(&self, f: F)
+  where
+    F: FnOnce(&mut v8::HandleScope) + Send + 'static,
+  {
+    self.tasks.spawn(Box::new(f))
+  }
+
+  /// Spawn a task that runs within the [`crate::JsRuntime`] event loop from a different thread
+  /// than the runtime is running on.
+  ///
+  /// This function will deadlock if called from the same thread as the [`crate::JsRuntime`], and
+  /// there are no checks for this case.
+  ///
+  /// As this function blocks until the task has run to completion (or panics/deadlocks), it is
+  /// safe to borrow data from the local environment and use it within the closure.
+  ///
+  /// The task is handed off to be run the next time the event loop is polled, and there are
+  /// no guarantees as to when this may happen, however the function will not return until the
+  /// task has been fully run to completion.
+  ///
+  /// # Safety
+  ///
+  /// The task shares the same [`v8::HandleScope`] as the core event loop, which means that it
+  /// must maintain the scope in a valid state to avoid corrupting or destroying the runtime.
+  ///
+  /// For example, if the code called by this task can raise an exception, the task must ensure
+  /// that it calls that code within a new [`v8::TryCatch`] to avoid the exception leaking to the
+  /// event loop's [`v8::HandleScope`].
+  pub fn spawn_blocking<'a, F, T>(&self, f: F) -> T
+  where
+    F: FnOnce(&mut v8::HandleScope) -> T + Send + 'a,
+    T: Send + 'a,
+  {
+    let (tx, rx) = std::sync::mpsc::sync_channel(0);
+    let task: Box<dyn FnOnce(&mut v8::HandleScope<'_>) + Send> =
+      Box::new(|scope| {
+        let r = f(scope);
+        _ = tx.send(r);
+      });
+    // SAFETY: We can safely transmute to the 'static lifetime because we guarantee this method will either
+    // complete fully by the time it returns, deadlock or panic.
+    let task: SendTask = unsafe { std::mem::transmute(task) };
+    self.tasks.spawn(task);
+    rx.recv().unwrap()
+  }
+}
+
+#[cfg(test)]
+mod tests {
+  use super::*;
+  use std::future::poll_fn;
+  use tokio::task::LocalSet;
+
+  #[test]
+  fn test_spawner_serial() {
+    let runtime = tokio::runtime::Builder::new_multi_thread()
+      .worker_threads(1)
+      .build()
+      .unwrap();
+    runtime.block_on(async {
+      let factory = Arc::<V8TaskSpawnerFactory>::default();
+      let cross_thread_spawner = factory.clone().new_cross_thread_spawner();
+      let local_set = LocalSet::new();
+
+      const COUNT: usize = 1000;
+
+      let task = runtime.spawn(async move {
+        for _ in 0..COUNT {
+          cross_thread_spawner.spawn(|_| {});
+        }
+      });
+
+      local_set.spawn_local(async move {
+        let mut count = 0;
+        loop {
+          count += poll_fn(|cx| factory.poll_inner(cx)).await.len();
+          if count >= COUNT {
+            break;
+          }
+        }
+      });
+
+      local_set.await;
+      _ = task.await;
+    });
+  }
+
+  #[test]
+  fn test_spawner_parallel() {
+    let runtime = tokio::runtime::Builder::new_multi_thread()
+      .worker_threads(1)
+      .build()
+      .unwrap();
+    runtime.block_on(async {
+      let factory = Arc::<V8TaskSpawnerFactory>::default();
+      let cross_thread_spawner = factory.clone().new_cross_thread_spawner();
+      let local_set = LocalSet::new();
+
+      const COUNT: usize = 100;
+      let mut tasks = vec![];
+      for _ in 0..COUNT {
+        let cross_thread_spawner = cross_thread_spawner.clone();
+        tasks.push(runtime.spawn(async move {
+          cross_thread_spawner.spawn(|_| {});
+        }));
+      }
+
+      local_set.spawn_local(async move {
+        let mut count = 0;
+        loop {
+          count += poll_fn(|cx| factory.poll_inner(cx)).await.len();
+          if count >= COUNT {
+            break;
+          }
+        }
+      });
+
+      local_set.await;
+      for task in tasks.drain(..) {
+        _ = task.await;
+      }
+    });
+  }
+}
diff --git a/ops/Cargo.toml b/ops/Cargo.toml
index 05a6a90..e956502 100644
--- a/ops/Cargo.toml
+++ b/ops/Cargo.toml
@@ -2,7 +2,7 @@
 
 [package]
 name = "deno_ops"
-version = "0.108.0"
+version = "0.109.0"
 authors.workspace = true
 edition.workspace = true
 license.workspace = true
diff --git a/serde_v8/Cargo.toml b/serde_v8/Cargo.toml
index 491ccaa..d92eead 100644
--- a/serde_v8/Cargo.toml
+++ b/serde_v8/Cargo.toml
@@ -2,7 +2,7 @@
 
 [package]
 name = "serde_v8"
-version = "0.141.0"
+version = "0.142.0"
 authors.workspace = true
 edition.workspace = true
 license.workspace = true


Test patch:
diff --git a/core/modules/tests.rs b/core/modules/tests.rs
index f1a0f36..af9d86b 100644
--- a/core/modules/tests.rs
+++ b/core/modules/tests.rs
@@ -20,6 +20,7 @@ use crate::ModuleType;
 use crate::ResolutionKind;
 use crate::RuntimeOptions;
 use crate::Snapshot;
+use anyhow::bail;
 use anyhow::Error;
 use deno_ops::op2;
 use futures::future::poll_fn;
@@ -502,13 +503,13 @@ fn test_json_module() {
 
   let receiver = runtime.mod_evaluate(mod_a);
   futures::executor::block_on(runtime.run_event_loop(false)).unwrap();
-  futures::executor::block_on(receiver).unwrap().unwrap();
+  futures::executor::block_on(receiver).unwrap();
 
   runtime.instantiate_module(mod_b).unwrap();
 
   let receiver = runtime.mod_evaluate(mod_b);
   futures::executor::block_on(runtime.run_event_loop(false)).unwrap();
-  futures::executor::block_on(receiver).unwrap().unwrap();
+  futures::executor::block_on(receiver).unwrap();
 }
 
 #[test]
@@ -1204,7 +1205,7 @@ fn import_meta_snapshot() {
     #[allow(clippy::let_underscore_future)]
     let eval_fut = runtime.mod_evaluate(main_id);
     futures::executor::block_on(runtime.run_event_loop(false)).unwrap();
-    futures::executor::block_on(eval_fut).unwrap().unwrap();
+    futures::executor::block_on(eval_fut).unwrap();
     runtime.snapshot()
   };
 
@@ -1327,3 +1328,47 @@ async fn no_duplicate_loads() {
   let _ = runtime.mod_evaluate(a_id);
   runtime.run_event_loop(false).await.unwrap();
 }
+
+#[test]
+fn import_meta_resolve_cb() {
+  fn import_meta_resolve_cb(
+    _loader: &dyn ModuleLoader,
+    specifier: String,
+    _referrer: String,
+  ) -> Result<ModuleSpecifier, Error> {
+    if specifier == "foo" {
+      return Ok(ModuleSpecifier::parse("foo:bar").unwrap());
+    }
+
+    if specifier == "./mod.js" {
+      return Ok(ModuleSpecifier::parse("file:///mod.js").unwrap());
+    }
+
+    bail!("unexpected")
+  }
+
+  let mut runtime = JsRuntime::new(RuntimeOptions {
+    import_meta_resolve_callback: Some(Box::new(import_meta_resolve_cb)),
+    ..Default::default()
+  });
+
+  let spec = ModuleSpecifier::parse("file:///test.js").unwrap();
+  let source = r#"
+  if (import.meta.resolve("foo") !== "foo") throw new Error();
+  if (import.meta.resolve("./mod.js") !== "file:///mod.js") throw new Error();
+  let caught = false;
+  try {
+    import.meta.resolve("boom!");
+  } catch (e) {
+    if (!(e instanceof TypeError)) throw new Error();
+  }
+  if (!caught) throw new Error();
+  "#
+  .to_string();
+  let a_id_fut = runtime.load_main_module(&spec, Some(source.into()));
+  let a_id = futures::executor::block_on(a_id_fut).unwrap();
+
+  #[allow(clippy::let_underscore_future)]
+  let _ = runtime.mod_evaluate(a_id);
+  futures::executor::block_on(runtime.run_event_loop(false)).unwrap();
+}
diff --git a/core/runtime/tests/misc.rs b/core/runtime/tests/misc.rs
index 1a23d89..a828fae 100644
--- a/core/runtime/tests/misc.rs
+++ b/core/runtime/tests/misc.rs
@@ -20,6 +20,7 @@ use std::sync::Arc;
 use std::task::Context;
 use std::task::Poll;
 use std::time::Duration;
+use std::time::Instant;
 use url::Url;
 
 #[test]
@@ -771,9 +772,8 @@ fn terminate_during_module_eval() {
   runtime.v8_isolate().terminate_execution();
 
   let mod_result =
-    futures::executor::block_on(runtime.mod_evaluate(module_id)).unwrap();
+    futures::executor::block_on(runtime.mod_evaluate(module_id)).unwrap_err();
   assert!(mod_result
-    .unwrap_err()
     .to_string()
     .contains("JavaScript execution has been terminated"));
 }
@@ -857,7 +857,8 @@ async fn test_set_promise_reject_callback_top_level_await() {
   static PROMISE_REJECT: AtomicUsize = AtomicUsize::new(0);
 
   #[op2(fast)]
-  fn op_promise_reject() -> Result<(), AnyError> {
+  fn op_promise_reject(promise_type: u32) -> Result<(), AnyError> {
+    eprintln!("{promise_type}");
     PROMISE_REJECT.fetch_add(1, Ordering::Relaxed);
     Ok(())
   }
@@ -869,7 +870,7 @@ async fn test_set_promise_reject_callback_top_level_await() {
     ascii_str!(
       r#"
   Deno.core.ops.op_set_promise_reject_callback((type, promise, reason) => {
-    Deno.core.ops.op_promise_reject();
+    Deno.core.ops.op_promise_reject(type);
   });
   throw new Error('top level throw');
   "#
@@ -888,7 +889,11 @@ async fn test_set_promise_reject_callback_top_level_await() {
     .unwrap();
   let receiver = runtime.mod_evaluate(id);
   runtime.run_event_loop(false).await.unwrap();
-  receiver.await.unwrap().unwrap_err();
+  let err = receiver.await.unwrap_err();
+  assert_eq!(
+    err.to_string(),
+    "Error: top level throw\n    at file:///main.js:5:9"
+  );
 
   assert_eq!(1, PROMISE_REJECT.load(Ordering::Relaxed));
 }
@@ -1072,3 +1077,103 @@ export const TEST = "foo";
     ..Default::default()
   });
 }
+
+fn create_spawner_runtime() -> JsRuntime {
+  let mut runtime = JsRuntime::new(RuntimeOptions {
+    ..Default::default()
+  });
+  runtime
+    .execute_script("main", ascii_str!("function f() { return 42; }"))
+    .unwrap();
+  runtime
+}
+
+fn call_i32_function(scope: &mut v8::HandleScope) -> i32 {
+  let ctx = scope.get_current_context();
+  let global = ctx.global(scope);
+  let key = v8::String::new_external_onebyte_static(scope, b"f")
+    .unwrap()
+    .into();
+  let f: v8::Local<'_, v8::Function> =
+    global.get(scope, key).unwrap().try_into().unwrap();
+  let recv = v8::undefined(scope).into();
+  let res: v8::Local<v8::Integer> =
+    f.call(scope, recv, &[]).unwrap().try_into().unwrap();
+  res.int32_value(scope).unwrap()
+}
+
+#[tokio::test]
+async fn task_spawner() {
+  let mut runtime = create_spawner_runtime();
+  let value = Arc::new(AtomicUsize::new(0));
+  let value_clone = value.clone();
+  runtime
+    .op_state()
+    .borrow()
+    .borrow::<V8TaskSpawner>()
+    .spawn(move |scope| {
+      let res = call_i32_function(scope);
+      value_clone.store(res as _, Ordering::SeqCst);
+    });
+  poll_fn(|cx| runtime.poll_event_loop(cx, false))
+    .await
+    .unwrap();
+  assert_eq!(value.load(Ordering::SeqCst), 42);
+}
+
+#[tokio::test]
+async fn task_spawner_cross_thread() {
+  let mut runtime = create_spawner_runtime();
+  let value = Arc::new(AtomicUsize::new(0));
+  let value_clone = value.clone();
+  let spawner = runtime
+    .op_state()
+    .borrow()
+    .borrow::<V8CrossThreadTaskSpawner>()
+    .clone();
+
+  std::thread::spawn(move || {
+    spawner.spawn(move |scope| {
+      let res = call_i32_function(scope);
+      value_clone.store(res as _, Ordering::SeqCst);
+    });
+  });
+
+  // Async spin while we wait for this to complete
+  let start = Instant::now();
+  while value.load(Ordering::SeqCst) != 42 {
+    poll_fn(|cx| runtime.poll_event_loop(cx, false))
+      .await
+      .unwrap();
+    tokio::time::sleep(Duration::from_millis(10)).await;
+    assert!(start.elapsed().as_secs() < 60);
+  }
+}
+
+#[tokio::test]
+async fn task_spawner_cross_thread_blocking() {
+  let mut runtime = create_spawner_runtime();
+
+  let value = Arc::new(AtomicUsize::new(0));
+  let value_clone = value.clone();
+  let spawner = runtime
+    .op_state()
+    .borrow()
+    .borrow::<V8CrossThreadTaskSpawner>()
+    .clone();
+
+  std::thread::spawn(move || {
+    let res = spawner.spawn_blocking(call_i32_function);
+    value_clone.store(res as _, Ordering::SeqCst);
+  });
+
+  // Async spin while we wait for this to complete
+  let start = Instant::now();
+  while value.load(Ordering::SeqCst) != 42 {
+    poll_fn(|cx| runtime.poll_event_loop(cx, false))
+      .await
+      .unwrap();
+    tokio::time::sleep(Duration::from_millis(10)).await;
+    assert!(start.elapsed().as_secs() < 60);
+  }
+}
diff --git a/core/runtime/tests/ops.rs b/core/runtime/tests/ops.rs
index f9fe2a2..e71ad55 100644
--- a/core/runtime/tests/ops.rs
+++ b/core/runtime/tests/ops.rs
@@ -550,23 +550,23 @@ await op_void_async_deferred();
     .unwrap();
   let mut rx = runtime.mod_evaluate(id);
 
-  let res = tokio::select! {
+  tokio::select! {
     // Not using biased mode leads to non-determinism for relatively simple
     // programs.
     biased;
 
     maybe_result = &mut rx => {
       debug!("received module evaluate {:#?}", maybe_result);
-      maybe_result.expect("Module evaluation result not provided.")
+      maybe_result
     }
 
     event_loop_result = runtime.run_event_loop(false) => {
       event_loop_result.unwrap();
-      let maybe_result = rx.await;
-      maybe_result.expect("Module evaluation result not provided.")
+
+      rx.await
     }
-  };
-  res.unwrap();
+  }
+  .expect("Failed to get module result");
 }
 
 #[op2(async)]