Run http_request on a tokio task

Author: coolreader18

crates/core/src/host/instance_env.rs

@@ -10,6 +10,7 @@ use crate::subscription::module_subscription_manager::{from_tx_offset, Transacti
 use crate::util::prometheus_handle::IntGaugeExt;
 use chrono::{DateTime, Utc};
 use core::mem;
+use futures::TryFutureExt;
 use parking_lot::{Mutex, MutexGuard};
 use smallvec::SmallVec;
 use spacetimedb_client_api_messages::energy::EnergyQuanta;
@@ -865,7 +866,8 @@ impl InstanceEnv {
         // TODO(perf): Stash a long-lived `Client` in the env somewhere, rather than building a new one for each call.
         let execute_fut = reqwest::Client::new().execute(reqwest);
 
-        let response_fut = async {
+        // Run the future that does IO work on a tokio worker thread, where it's more efficent.
+        let response_fut = tokio::spawn(async {
             // `reqwest::Error` may contain sensitive info, namely the full URL with query params.
             // We'll strip those with `strip_query_params_from_eqwest_error`
             // after `await`ing `response_fut` below.
@@ -880,7 +882,8 @@ impl InstanceEnv {
             let body = http_body_util::BodyExt::collect(body).await?.to_bytes();
 
             Ok((response, body))
-        };
+        })
+        .unwrap_or_else(|e| std::panic::resume_unwind(e.into_panic()));
 
         let database_identity = *self.database_identity();
 

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

@@ -415,13 +415,10 @@ impl JsInstance {
     }
 
     pub async fn call_procedure(&mut self, params: CallProcedureParams) -> CallProcedureReturn {
-        // Get a handle to the current tokio runtime, and pass it to the worker
-        // so that it can execute futures.
-        let rt = tokio::runtime::Handle::current();
         *self
             .send_recv(
                 JsWorkerReply::into_call_procedure,
-                JsWorkerRequest::CallProcedure { params, rt },
+                JsWorkerRequest::CallProcedure { params },
             )
             .await
     }
@@ -436,12 +433,9 @@ impl JsInstance {
         &mut self,
         params: ScheduledFunctionParams,
     ) -> CallScheduledFunctionResult {
-        // Get a handle to the current tokio runtime, and pass it to the worker
-        // so that it can execute futures.
-        let rt = tokio::runtime::Handle::current();
         self.send_recv(
             JsWorkerReply::into_call_scheduled_function,
-            JsWorkerRequest::CallScheduledFunction(params, rt),
+            JsWorkerRequest::CallScheduledFunction(params),
         )
         .await
     }
@@ -481,10 +475,7 @@ enum JsWorkerRequest {
     /// See [`JsInstance::call_view`].
     CallView { cmd: ViewCommand },
     /// See [`JsInstance::call_procedure`].
-    CallProcedure {
-        params: CallProcedureParams,
-        rt: tokio::runtime::Handle,
-    },
+    CallProcedure { params: CallProcedureParams },
     /// See [`JsInstance::clear_all_clients`].
     ClearAllClients,
     /// See [`JsInstance::call_identity_connected`].
@@ -496,7 +487,7 @@ enum JsWorkerRequest {
     /// See [`JsInstance::init_database`].
     InitDatabase(Program),
     /// See [`JsInstance::call_scheduled_function`].
-    CallScheduledFunction(ScheduledFunctionParams, tokio::runtime::Handle),
+    CallScheduledFunction(ScheduledFunctionParams),
 }
 
 /// Performs some of the startup work of [`spawn_instance_worker`].
@@ -568,6 +559,8 @@ async fn spawn_instance_worker(
         let _guard = load_balance_guard;
         core_pinner.pin_now();
 
+        let _entered = rt.enter();
+
         // Create the isolate and scope.
         let mut isolate = new_isolate();
         scope_with_context!(let scope, &mut isolate, Context::new(scope, Default::default()));
@@ -651,11 +644,7 @@ async fn spawn_instance_worker(
                     let (res, trapped) = instance_common.handle_cmd(cmd, &mut inst);
                     reply("call_view", JsWorkerReply::CallView(res.into()), trapped);
                 }
-                JsWorkerRequest::CallProcedure { params, rt } => {
-                    // The callee passed us a handle to their tokio runtime - enter its
-                    // context so that we can execute futures.
-                    let _guard = rt.enter();
-
+                JsWorkerRequest::CallProcedure { params } => {
                     let (res, trapped) = instance_common
                         .call_procedure(params, &mut inst)
                         .now_or_never()
@@ -699,9 +688,7 @@ async fn spawn_instance_worker(
                         init_database(replica_ctx, &module_common.info().module_def, program, call_reducer);
                     reply("init_database", InitDatabase(res), trapped);
                 }
-                JsWorkerRequest::CallScheduledFunction(params, rt) => {
-                    let _guard = rt.enter();
-
+                JsWorkerRequest::CallScheduledFunction(params) => {
                     let (res, trapped) = instance_common
                         .call_scheduled_function(params, &mut inst)
                         .now_or_never()