minimize diff

Author: joshua-spacetime

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

@@ -333,7 +333,7 @@ impl JsInstance {
         Ok(value)
     }
 
-    async fn run_on_thread_inner<F, R>(&self, f: F) -> Result<R, WorkerDisconnected>
+    pub async fn run_on_thread<F, R>(&self, f: F) -> R
     where
         F: AsyncFnOnce() -> R + Send + 'static,
         R: Send + 'static,
@@ -352,22 +352,15 @@ impl JsInstance {
                 .boxed_local()
             })))
             .await
-            .map_err(|_| WorkerDisconnected)?;
+            .unwrap_or_else(|_| panic!("worker should stay live while handling {}", type_name::<R>()));
 
-        Ok(match rx.await.map_err(|_| WorkerDisconnected)? {
-            Ok(r) => r,
-            Err(e) => std::panic::resume_unwind(e),
-        })
-    }
-
-    pub async fn run_on_thread<F, R>(&self, f: F) -> R
-    where
-        F: AsyncFnOnce() -> R + Send + 'static,
-        R: Send + 'static,
-    {
-        self.run_on_thread_inner(f)
+        match rx
             .await
             .unwrap_or_else(|_| panic!("worker should stay live while handling {}", type_name::<R>()))
+        {
+            Ok(r) => r,
+            Err(e) => std::panic::resume_unwind(e),
+        }
     }
 
     pub async fn update_database(
@@ -386,13 +379,8 @@ impl JsInstance {
         .unwrap_or_else(|_| panic!("worker should stay live while updating the database"))
     }
 
-    async fn call_reducer_inner(&self, params: CallReducerParams) -> Result<ReducerCallResult, WorkerDisconnected> {
-        self.send_request(|reply_tx| JsWorkerRequest::CallReducer { reply_tx, params })
-            .await
-    }
-
     pub async fn call_reducer(&self, params: CallReducerParams) -> ReducerCallResult {
-        self.call_reducer_inner(params)
+        self.send_request(|reply_tx| JsWorkerRequest::CallReducer { reply_tx, params })
             .await
             .unwrap_or_else(|_| panic!("worker should stay live while calling a reducer"))
     }
@@ -403,62 +391,36 @@ impl JsInstance {
             .unwrap_or_else(|_| panic!("worker should stay live while clearing clients"))
     }
 
-    async fn call_identity_connected_inner(
+    pub async fn call_identity_connected(
         &self,
         caller_auth: ConnectionAuthCtx,
         caller_connection_id: ConnectionId,
-    ) -> Result<Result<(), ClientConnectedError>, WorkerDisconnected> {
+    ) -> Result<(), ClientConnectedError> {
         self.send_request(|reply_tx| JsWorkerRequest::CallIdentityConnected {
             reply_tx,
             caller_auth,
             caller_connection_id,
         })
         .await
+        .unwrap_or_else(|_| panic!("worker should stay live while running client_connected"))
     }
 
-    pub async fn call_identity_connected(
-        &self,
-        caller_auth: ConnectionAuthCtx,
-        caller_connection_id: ConnectionId,
-    ) -> Result<(), ClientConnectedError> {
-        self.call_identity_connected_inner(caller_auth, caller_connection_id)
-            .await
-            .unwrap_or_else(|_| panic!("worker should stay live while running client_connected"))
-    }
-
-    async fn call_identity_disconnected_inner(
+    pub async fn call_identity_disconnected(
         &self,
         caller_identity: Identity,
         caller_connection_id: ConnectionId,
-    ) -> Result<Result<(), ReducerCallError>, WorkerDisconnected> {
+    ) -> Result<(), ReducerCallError> {
         self.send_request(|reply_tx| JsWorkerRequest::CallIdentityDisconnected {
             reply_tx,
             caller_identity,
             caller_connection_id,
         })
         .await
-    }
-
-    pub async fn call_identity_disconnected(
-        &self,
-        caller_identity: Identity,
-        caller_connection_id: ConnectionId,
-    ) -> Result<(), ReducerCallError> {
-        self.call_identity_disconnected_inner(caller_identity, caller_connection_id)
-            .await
-            .unwrap_or_else(|_| panic!("worker should stay live while running client_disconnected"))
-    }
-
-    async fn disconnect_client_inner(
-        &self,
-        client_id: ClientActorId,
-    ) -> Result<Result<(), ReducerCallError>, WorkerDisconnected> {
-        self.send_request(|reply_tx| JsWorkerRequest::DisconnectClient { reply_tx, client_id })
-            .await
+        .unwrap_or_else(|_| panic!("worker should stay live while running client_disconnected"))
     }
 
     pub async fn disconnect_client(&self, client_id: ClientActorId) -> Result<(), ReducerCallError> {
-        self.disconnect_client_inner(client_id)
+        self.send_request(|reply_tx| JsWorkerRequest::DisconnectClient { reply_tx, client_id })
             .await
             .unwrap_or_else(|_| panic!("worker should stay live while disconnecting a client"))
     }
@@ -475,13 +437,8 @@ impl JsInstance {
             .unwrap_or_else(|_| panic!("worker should stay live while calling a procedure"))
     }
 
-    async fn call_view_inner(&self, cmd: ViewCommand) -> Result<ViewCommandResult, WorkerDisconnected> {
-        self.send_request(|reply_tx| JsWorkerRequest::CallView { reply_tx, cmd })
-            .await
-    }
-
     pub async fn call_view(&self, cmd: ViewCommand) -> ViewCommandResult {
-        self.call_view_inner(cmd)
+        self.send_request(|reply_tx| JsWorkerRequest::CallView { reply_tx, cmd })
             .await
             .unwrap_or_else(|_| panic!("worker should stay live while calling a view"))
     }
@@ -678,7 +635,25 @@ impl JsReducerLane {
     {
         let span = tracing::Span::current();
         self.run_nonreplayable("run_on_thread", async move |inst| {
-            inst.run_on_thread_inner(async move || f().instrument(span).await).await
+            let (tx, rx) = oneshot::channel();
+
+            inst.request_tx
+                .send_async(JsWorkerRequest::RunFunction(Box::new(move || {
+                    async move {
+                        let result = AssertUnwindSafe(f().instrument(span)).catch_unwind().await;
+                        if let Err(Err(_panic)) = tx.send(result) {
+                            tracing::warn!("uncaught panic on `SingleCoreExecutor`")
+                        }
+                    }
+                    .boxed_local()
+                })))
+                .await
+                .map_err(|_| WorkerDisconnected)?;
+
+            Ok(match rx.await.map_err(|_| WorkerDisconnected)? {
+                Ok(r) => r,
+                Err(e) => std::panic::resume_unwind(e),
+            })
         })
         .await
         .map_err(|_| anyhow::anyhow!("reducer worker exited while running a non-replayable module-thread task"))
@@ -708,7 +683,8 @@ impl JsReducerLane {
 
     pub async fn call_reducer(&self, params: CallReducerParams) -> ReducerCallResult {
         self.run_replayable(params, "call_reducer", |inst, params| async move {
-            inst.call_reducer_inner(params).await
+            inst.send_request(|reply_tx| JsWorkerRequest::CallReducer { reply_tx, params })
+                .await
         })
         .await
     }
@@ -729,8 +705,12 @@ impl JsReducerLane {
             (caller_auth, caller_connection_id),
             "call_identity_connected",
             |inst, (caller_auth, caller_connection_id)| async move {
-                inst.call_identity_connected_inner(caller_auth, caller_connection_id)
-                    .await
+                inst.send_request(|reply_tx| JsWorkerRequest::CallIdentityConnected {
+                    reply_tx,
+                    caller_auth,
+                    caller_connection_id,
+                })
+                .await
             },
         )
         .await
@@ -745,16 +725,21 @@ impl JsReducerLane {
             (caller_identity, caller_connection_id),
             "call_identity_disconnected",
             |inst, (caller_identity, caller_connection_id)| async move {
-                inst.call_identity_disconnected_inner(caller_identity, caller_connection_id)
-                    .await
+                inst.send_request(|reply_tx| JsWorkerRequest::CallIdentityDisconnected {
+                    reply_tx,
+                    caller_identity,
+                    caller_connection_id,
+                })
+                .await
             },
         )
         .await
     }
 
     pub async fn disconnect_client(&self, client_id: ClientActorId) -> Result<(), ReducerCallError> {
         self.run_replayable(client_id, "disconnect_client", |inst, client_id| async move {
-            inst.disconnect_client_inner(client_id).await
+            inst.send_request(|reply_tx| JsWorkerRequest::DisconnectClient { reply_tx, client_id })
+                .await
         })
         .await
     }
@@ -772,9 +757,12 @@ impl JsReducerLane {
         // the websocket request payload types they carry are not cloneable.
         // If a worker dies here we surface the failure rather than silently
         // re-running a command that may already have produced side effects.
-        self.run_nonreplayable("call_view", async move |inst| inst.call_view_inner(cmd).await)
-            .await
-            .unwrap_or_else(|_| panic!("reducer worker exited while handling a non-replayable view command"))
+        self.run_nonreplayable("call_view", async move |inst| {
+            inst.send_request(|reply_tx| JsWorkerRequest::CallView { reply_tx, cmd })
+                .await
+        })
+        .await
+        .unwrap_or_else(|_| panic!("reducer worker exited while handling a non-replayable view command"))
     }
 
     pub(in crate::host) async fn call_scheduled_function(