Fix panic kills running engine in query-engine-tests

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/Cargo.toml

@@ -12,7 +12,7 @@ request-handlers = { path = "../../request-handlers" }
 tokio.workspace = true
 query-core = { path = "../../core", features = ["metrics"] }
 sql-query-connector = { path = "../../connectors/sql-query-connector" }
-query-engine = { path = "../../query-engine"}
+query-engine = { path = "../../query-engine" }
 psl.workspace = true
 user-facing-errors = { path = "../../../libs/user-facing-errors" }
 thiserror = "1.0"
@@ -30,9 +30,10 @@ indoc.workspace = true
 enumflags2 = "0.7"
 hyper = { version = "0.14", features = ["full"] }
 indexmap = { version = "1.0", features = ["serde-1"] }
-query-engine-metrics = {path = "../../metrics"}
+query-engine-metrics = { path = "../../metrics" }
 quaint.workspace = true
 jsonrpc-core = "17"
+insta = "1.7.1"
 
 # Only this version is vetted, upgrade only after going through the code,
 # as this is a small crate with little user base.

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

@@ -287,7 +287,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