V8: Let a worker thread own the isolate (#3401)

# Description of Changes

1. Moves work of `JsInstance` to a worker thread to avoid re-making the
isolate all the time.
2. Does some minor refactoring.
3. Deals with `JsInstanceEnv` being potentially unset.

Fixes #3460.

# API and ABI breaking changes

None

# Expected complexity level and risk

3

# Testing

Should be covered by exiting tests.

Author: Centril

crates/core/src/host/host_controller.rs

@@ -603,14 +603,14 @@ async fn make_module_host(
         let start = Instant::now();
         let module_host = match host_type {
             HostType::Wasm => {
-                let actor = runtimes.wasmtime.make_actor(mcc)?;
+                let (actor, init_inst) = runtimes.wasmtime.make_actor(mcc)?;
                 trace!("wasmtime::make_actor blocked for {:?}", start.elapsed());
-                ModuleHost::new(actor, unregister, executor, database_identity)
+                ModuleHost::new(actor, init_inst, unregister, executor, database_identity)
             }
             HostType::Js => {
-                let actor = runtimes.v8.make_actor(mcc)?;
+                let (actor, init_inst) = runtimes.v8.make_actor(mcc)?;
                 trace!("v8::make_actor blocked for {:?}", start.elapsed());
-                ModuleHost::new(actor, unregister, executor, database_identity)
+                ModuleHost::new(actor, init_inst, unregister, executor, database_identity)
             }
         };
         Ok((program, module_host))

crates/core/src/host/module_common.rs

@@ -8,7 +8,7 @@ use crate::{
         wasm_common::{module_host_actor::DescribeError, DESCRIBE_MODULE_DUNDER},
         Scheduler,
     },
-    module_host_context::ModuleCreationContext,
+    module_host_context::ModuleCreationContextLimited,
     replica_context::ReplicaContext,
 };
 use spacetimedb_lib::{Identity, RawModuleDef};
@@ -17,7 +17,7 @@ use std::sync::Arc;
 
 /// Builds a [`ModuleCommon`] from a [`RawModuleDef`].
 pub fn build_common_module_from_raw(
-    mcc: ModuleCreationContext,
+    mcc: ModuleCreationContextLimited,
     raw_def: RawModuleDef,
 ) -> Result<ModuleCommon, ValidationErrors> {
     // Perform a bunch of validation on the raw definition.
@@ -31,7 +31,7 @@ pub fn build_common_module_from_raw(
         def,
         replica_ctx.owner_identity,
         replica_ctx.database_identity,
-        mcc.program.hash,
+        mcc.program_hash,
         log_tx,
         replica_ctx.subscriptions.clone(),
     );

crates/core/src/host/module_host.rs

@@ -321,7 +321,9 @@ impl ReducersMap {
 /// A runtime that can create modules.
 pub trait ModuleRuntime {
     /// Creates a module based on the context `mcc`.
-    fn make_actor(&self, mcc: ModuleCreationContext<'_>) -> anyhow::Result<Module>;
+    ///
+    /// Also returns the initial instance for the module.
+    fn make_actor(&self, mcc: ModuleCreationContext<'_>) -> anyhow::Result<(Module, Instance)>;
 }
 
 pub enum Module {
@@ -357,10 +359,10 @@ impl Module {
             Module::Js(module) => module.info(),
         }
     }
-    fn create_instance(&self) -> Instance {
+    async fn create_instance(&self) -> Instance {
         match self {
             Module::Wasm(module) => Instance::Wasm(Box::new(module.create_instance())),
-            Module::Js(module) => Instance::Js(Box::new(module.create_instance())),
+            Module::Js(module) => Instance::Js(Box::new(module.create_instance().await)),
         }
     }
     fn host_type(&self) -> HostType {
@@ -554,7 +556,7 @@ impl CreateInstanceTimeMetric {
 }
 
 impl ModuleInstanceManager {
-    fn new(module: Arc<Module>, database_identity: Identity) -> Self {
+    fn new(module: Arc<Module>, init_inst: Instance, database_identity: Identity) -> Self {
         let host_type = module.host_type();
         let create_instance_time_metric = CreateInstanceTimeMetric {
             metric: WORKER_METRICS
@@ -563,19 +565,24 @@ impl ModuleInstanceManager {
             host_type,
             database_identity,
         };
+
+        // Add the first instance.
+        let mut instances = VecDeque::new();
+        instances.push_front(init_inst);
+
         Self {
-            instances: Default::default(),
+            instances,
             module,
             create_instance_time_metric,
         }
     }
-    fn get_instance(&mut self) -> Instance {
+    async fn get_instance(&mut self) -> Instance {
         if let Some(inst) = self.instances.pop_back() {
             inst
         } else {
             let start_time = std::time::Instant::now();
             // TODO: should we be calling `create_instance` on the `SingleCoreExecutor` rather than the calling thread?
-            let res = self.module.create_instance();
+            let res = self.module.create_instance().await;
             let elapsed_time = start_time.elapsed();
             self.create_instance_time_metric.observe(elapsed_time);
             res
@@ -690,6 +697,7 @@ pub enum ClientConnectedError {
 impl ModuleHost {
     pub(super) fn new(
         module: Module,
+        init_inst: Instance,
         on_panic: impl Fn() + Send + Sync + 'static,
         executor: SingleCoreExecutor,
         database_identity: Identity,
@@ -700,7 +708,8 @@ impl ModuleHost {
 
         let module_clone = module.clone();
 
-        let instance_manager = Arc::new(Mutex::new(ModuleInstanceManager::new(module_clone, database_identity)));
+        let instance_manager = ModuleInstanceManager::new(module_clone, init_inst, database_identity);
+        let instance_manager = Arc::new(Mutex::new(instance_manager));
 
         ModuleHost {
             info,
@@ -807,7 +816,7 @@ impl ModuleHost {
             (self.on_panic)();
         });
 
-        let mut instance = self.instance_manager.lock().await.get_instance();
+        let mut instance = self.instance_manager.lock().await.get_instance().await;
 
         let (res, instance) = self
             .executor

crates/core/src/host/v8/budget.rs

@@ -1,3 +1,5 @@
+#![allow(dead_code)]
+
 //! Provides budget, energy, timeout, and long-running logging facilities.
 //!
 //! These are all driven by [`with_timeout_and_cb_every`] for V8 modules
@@ -47,7 +49,11 @@ pub(super) type InterruptCallback = extern "C" fn(&mut Isolate, *mut c_void);
 /// every [`EPOCH_TICKS_PER_SECOND`] ticks (~every 1 second)
 /// to log that the reducer is still running.
 pub(super) extern "C" fn cb_log_long_running(isolate: &mut Isolate, _: *mut c_void) {
-    let env = env_on_isolate(isolate);
+    let Some(env) = env_on_isolate(isolate) else {
+        // All we can do is log something.
+        tracing::error!("`JsInstanceEnv` not set");
+        return;
+    };
     let database = env.instance_env.replica_ctx.database_identity;
     let reducer = env.reducer_name();
     let dur = env.reducer_start().elapsed();
@@ -61,7 +67,6 @@ pub(super) extern "C" fn cb_noop(_: &mut Isolate, _: *mut c_void) {}
 /// when `budget` has been used up.
 ///
 /// Every `callback_every` ticks, `callback` is called.
-#[allow(dead_code)]
 fn run_timeout_and_cb_every(
     handle: IsolateHandle,
     callback_every: u64,

crates/core/src/host/v8/error.rs

@@ -402,10 +402,33 @@ pub(super) fn log_traceback(func_type: &str, func: &str, e: &anyhow::Error) {
 pub(super) fn catch_exception<'scope, T>(
     scope: &mut PinScope<'scope, '_>,
     body: impl FnOnce(&mut PinScope<'scope, '_>) -> Result<T, ErrorOrException<ExceptionThrown>>,
-) -> Result<T, ErrorOrException<JsError>> {
+) -> Result<T, (ErrorOrException<JsError>, CanContinue)> {
     tc_scope!(scope, scope);
     body(scope).map_err(|e| match e {
-        ErrorOrException::Err(e) => ErrorOrException::Err(e),
-        ErrorOrException::Exception(_) => ErrorOrException::Exception(JsError::from_caught(scope)),
+        ErrorOrException::Err(e) => (ErrorOrException::Err(e), CanContinue::Yes),
+        ErrorOrException::Exception(_) => {
+            let error = ErrorOrException::Exception(JsError::from_caught(scope));
+
+            let can_continue = if scope.can_continue() {
+                // We can continue.
+                CanContinue::Yes
+            } else if scope.has_terminated() {
+                // We can continue if we do `Isolate::cancel_terminate_execution`.
+                CanContinue::YesCancelTermination
+            } else {
+                // We cannot.
+                CanContinue::No
+            };
+
+            (error, can_continue)
+        }
     })
 }
+
+/// Encodes whether it is safe to continue using the [`Isolate`]
+/// for further execution after [`catch_exception`] has happened.
+pub(super) enum CanContinue {
+    Yes,
+    YesCancelTermination,
+    No,
+}