Fix panic kills running engine in query-engine-tests (#4499)

* I didn't have wasm-pack installed, let's fix that

* Update wasm-bindgen-futures to 0.4.39

This includes rustwasm/wasm-bindgen#3203 that use queueMicrotask
to transalate spawn_local rust code.

This has fixed rustwasm/wasm-bindgen#2392 which was an issue about not being able to catch async wasm traps. This might (or not) have an effect on the issue we are trying to solve in here.

* Revert "Update wasm-bindgen-futures to 0.4.39"

This reverts commit 9a494dc.

* Restart executor when it dies

* Document Restartable

* Remove async_panic_to_js_error in WASM query engine

* Rename p -> process

* Use tokio::sync::RwLock rather than futures::lock::Mutex

* Better error messaging

* Fixing clippy

* Exclude unit tests for wasm32 when compiling the binary for other architectures

libs/crosstarget-utils/src/common.rs

@@ -1,13 +1,7 @@
 use std::fmt::Display;
 
 #[derive(Debug)]
-pub struct SpawnError {}
-
-impl SpawnError {
-    pub fn new() -> Self {
-        SpawnError {}
-    }
-}
+pub struct SpawnError;
 
 impl Display for SpawnError {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
@@ -18,13 +12,7 @@ impl Display for SpawnError {
 impl std::error::Error for SpawnError {}
 
 #[derive(Debug)]
-pub struct TimeoutError {}
-
-impl TimeoutError {
-    pub fn new() -> Self {
-        TimeoutError {}
-    }
-}
+pub struct TimeoutError;
 
 impl Display for TimeoutError {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {

libs/crosstarget-utils/src/native/spawn.rs

@@ -7,5 +7,5 @@ where
     F: Future + 'static + Send,
     F::Output: Send + 'static,
 {
-    tokio::spawn(future).await.map_err(|_| SpawnError::new())
+    tokio::spawn(future).await.map_err(|_| SpawnError)
 }

libs/crosstarget-utils/src/native/time.rs

@@ -21,7 +21,7 @@ impl ElapsedTimeCounter {
     }
 }
 
-pub async fn sleep(duration: Duration) -> () {
+pub async fn sleep(duration: Duration) {
     tokio::time::sleep(duration).await
 }
 
@@ -31,5 +31,5 @@ where
 {
     let result = tokio::time::timeout(duration, future).await;
 
-    result.map_err(|_| TimeoutError::new())
+    result.map_err(|_| TimeoutError)
 }

libs/crosstarget-utils/src/wasm/time.rs

@@ -50,7 +50,7 @@ where
 {
     tokio::select! {
         result = future => Ok(result),
-        _ = sleep(duration) => Err(TimeoutError::new())
+        _ = sleep(duration) => Err(TimeoutError)
     }
 }
 

query-engine/connector-test-kit-rs/query-tests-setup/src/connector_tag/js.rs

@@ -3,7 +3,7 @@ mod external_process;
 use super::*;
 use external_process::*;
 use serde::de::DeserializeOwned;
-use std::{collections::HashMap, sync::atomic::AtomicU64};
+use std::sync::atomic::AtomicU64;
 use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
 
 pub(crate) async fn executor_process_request<T: DeserializeOwned>(

query-engine/connector-test-kit-rs/query-tests-setup/src/connector_tag/js/external_process.rs

@@ -1,8 +1,13 @@
 use super::*;
 use once_cell::sync::Lazy;
 use serde::de::DeserializeOwned;
-use std::{fmt::Display, io::Write as _, sync::atomic::Ordering};
-use tokio::sync::{mpsc, oneshot};
+use std::{
+    error::Error as StdError,
+    fmt::Display,
+    io::Write as _,
+    sync::{atomic::Ordering, Arc},
+};
+use tokio::sync::{mpsc, oneshot, RwLock};
 
 type Result<T> = std::result::Result<T, Box<dyn std::error::Error + Send + Sync>>;
 
@@ -29,6 +34,17 @@ fn exit_with_message(status_code: i32, message: &str) -> ! {
 }
 
 impl ExecutorProcess {
+    fn spawn() -> ExecutorProcess {
+        match std::thread::spawn(ExecutorProcess::new).join() {
+            Ok(Ok(process)) => process,
+            Ok(Err(err)) => exit_with_message(1, &format!("Failed to start node process. Details: {err}")),
+            Err(err) => {
+                let err = err.downcast_ref::<String>().map(ToOwned::to_owned).unwrap_or_default();
+                exit_with_message(1, &format!("Panic while trying to start node process.\nDetails: {err}"))
+            }
+        }
+    }
+
     fn new() -> Result<ExecutorProcess> {
         let (sender, receiver) = mpsc::channel::<ReqImpl>(300);
 
@@ -81,15 +97,50 @@ impl ExecutorProcess {
     }
 }
 
-pub(super) static EXTERNAL_PROCESS: Lazy<ExecutorProcess> =
-    Lazy::new(|| match std::thread::spawn(ExecutorProcess::new).join() {
-        Ok(Ok(process)) => process,
-        Ok(Err(err)) => exit_with_message(1, &format!("Failed to start node process. Details: {err}")),
-        Err(err) => {
-            let err = err.downcast_ref::<String>().map(ToOwned::to_owned).unwrap_or_default();
-            exit_with_message(1, &format!("Panic while trying to start node process.\nDetails: {err}"))
+/// Wraps an ExecutorProcess allowing for restarting it.
+///
+/// A node process can die for a number of reasons, being one that any `panic!` occurring in Rust
+/// asynchronous code are translated to an abort trap by wasm-bindgen, which kills the node process.
+#[derive(Clone)]
+pub(crate) struct RestartableExecutorProcess {
+    process: Arc<RwLock<ExecutorProcess>>,
+}
+
+impl RestartableExecutorProcess {
+    fn new() -> Self {
+        Self {
+            process: Arc::new(RwLock::new(ExecutorProcess::spawn())),
         }
-    });
+    }
+
+    async fn restart(&self) {
+        let mut process = self.process.write().await;
+        *process = ExecutorProcess::spawn();
+    }
+
+    pub(crate) async fn request<T: DeserializeOwned>(&self, method: &str, params: serde_json::Value) -> Result<T> {
+        let p = self.process.read().await;
+        p.request(method, params).await
+    }
+}
+
+struct ExecutorProcessDiedError;
+
+impl fmt::Debug for ExecutorProcessDiedError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "The external test executor process died")
+    }
+}
+
+impl Display for ExecutorProcessDiedError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        fmt::Debug::fmt(self, f)
+    }
+}
+
+impl StdError for ExecutorProcessDiedError {}
+
+pub(super) static EXTERNAL_PROCESS: Lazy<RestartableExecutorProcess> = Lazy::new(RestartableExecutorProcess::new);
 
 type ReqImpl = (
     jsonrpc_core::MethodCall,
@@ -122,8 +173,7 @@ fn start_rpc_thread(mut receiver: mpsc::Receiver<ReqImpl>) -> Result<()> {
 
             let mut stdout = BufReader::new(process.stdout.unwrap()).lines();
             let mut stdin = process.stdin.unwrap();
-            let mut pending_requests: HashMap<jsonrpc_core::Id, oneshot::Sender<Result<serde_json::value::Value>>> =
-                HashMap::new();
+            let mut last_pending_request: Option<(jsonrpc_core::Id, oneshot::Sender<Result<serde_json::value::Value>>)> = None;
 
             loop {
                 tokio::select! {
@@ -137,7 +187,11 @@ fn start_rpc_thread(mut receiver: mpsc::Receiver<ReqImpl>) -> Result<()> {
                             {
                                 match serde_json::from_str::<jsonrpc_core::Output>(&line) {
                                     Ok(response) => {
-                                        let sender = pending_requests.remove(response.id()).unwrap();
+                                        let (id, sender) = last_pending_request.take().expect("got a response from the external process, but there was no pending request");
+                                        if &id != response.id() {
+                                            unreachable!("got a response from the external process, but the id didn't match. Are you running with cargo tests with `--test-threads=1`");
+                                        }
+
                                         match response {
                                             jsonrpc_core::Output::Success(success) => {
                                                 // The other end may be dropped if the whole
@@ -159,7 +213,12 @@ fn start_rpc_thread(mut receiver: mpsc::Receiver<ReqImpl>) -> Result<()> {
                             }
                             Ok(None) => // end of the stream
                             {
-                                exit_with_message(1, "child node process stdout closed")
+                                tracing::error!("Error when reading from child node process. Process might have exited. Restarting...");
+                                if let Some((_, sender)) = last_pending_request.take() {
+                                    sender.send(Err(Box::new(ExecutorProcessDiedError))).unwrap();
+                                }
+                                EXTERNAL_PROCESS.restart().await;
+                                break;
                             }
                             Err(err) => // log it
                             {
@@ -174,7 +233,7 @@ fn start_rpc_thread(mut receiver: mpsc::Receiver<ReqImpl>) -> Result<()> {
                                 exit_with_message(1, "The json-rpc client channel was closed");
                             }
                             Some((request, response_sender)) => {
-                                pending_requests.insert(request.id.clone(), response_sender);
+                                last_pending_request = Some((request.id.clone(), response_sender));
                                 let mut req = serde_json::to_vec(&request).unwrap();
                                 req.push(b'\n');
                                 stdin.write_all(&req).await.unwrap();

query-engine/connector-test-kit-rs/query-tests-setup/src/lib.rs

@@ -289,7 +289,9 @@ fn run_connector_test_impl(
             .unwrap();
             let schema_id = runner.schema_id();
 
-            test_fn(runner).await.unwrap();
+            if let Err(err) = test_fn(runner).await {
+                panic!("💥 Test failed due to an error: {err:?}");
+            }
 
             crate::teardown_project(&datamodel, db_schemas, schema_id)
                 .await

query-engine/driver-adapters/tests/wasm.rs

@@ -1,6 +1,6 @@
+#![cfg(target_os = "wasm32")]
 use wasm_bindgen_test::*;
 
-// use driver_adapters::types::ColumnType;
 use serde::{Deserialize, Serialize};
 use serde_repr::{Deserialize_repr, Serialize_repr};
 use tsify::Tsify;

query-engine/query-engine-wasm/Cargo.toml

@@ -6,7 +6,7 @@ edition = "2021"
 [lib]
 doc = false
 crate-type = ["cdylib"]
-name = "query_engine_wasm"
+name = "query_engine"
 
 [dependencies]
 

query-engine/query-engine-wasm/build.sh

@@ -21,6 +21,13 @@ else
     BUILD_PROFILE="--dev"
 fi
 
+# Check if wasm-pack is installed
+if ! command -v wasm-pack &> /dev/null
+then
+    echo "wasm-pack could not be found, installing now..."
+    # Install wasm-pack
+    curl https://rustwasm.github.io/wasm-pack/installer/init.sh -sSf | sh
+fi
 wasm-pack build $BUILD_PROFILE --target $OUT_TARGET
 
 sed -i '' 's/name = "query_engine"/name = "query_engine_wasm"/g' Cargo.toml