fix(native): Handle JsAsyncChannel error (without panic) (#7729)

Author: ovr

packages/cubejs-backend-native/src/channel.rs

@@ -1,23 +1,33 @@
 use std::cell::RefCell;
 use std::collections::HashMap;
+#[cfg(debug_assertions)]
+use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::Arc;
 
 use crate::transport::MapCubeErrExt;
 use async_trait::async_trait;
 use cubesql::transport::{SqlGenerator, SqlTemplates};
 use cubesql::CubeError;
-#[cfg(build = "debug")]
+#[cfg(debug_assertions)]
 use log::trace;
 use neon::prelude::*;
 use tokio::sync::oneshot;
 
 use crate::utils::bind_method;
 
-type JsAsyncStringChannelCallback = Box<dyn FnOnce(Result<String, CubeError>) + Send>;
-type JsAsyncChannelCallback =
-    Box<dyn FnOnce(&mut FunctionContext, Result<Handle<JsValue>, CubeError>) + Send>;
+type JsAsyncStringChannelCallback =
+    Box<dyn FnOnce(Result<String, CubeError>) -> Result<(), CubeError> + Send>;
+type JsAsyncChannelCallback = Box<
+    dyn FnOnce(&mut FunctionContext, Result<Handle<JsValue>, CubeError>) -> Result<(), CubeError>
+        + Send,
+>;
+
+#[cfg(debug_assertions)]
+static JS_ASYNC_CHANNEL_DEBUG_ID_SEQ: AtomicU64 = AtomicU64::new(0);
 
 pub struct JsAsyncChannel {
+    #[cfg(debug_assertions)]
+    _id: u64,
     callback: Option<JsAsyncChannelCallback>,
 }
 
@@ -26,32 +36,39 @@ type BoxedChannel = JsBox<RefCell<JsAsyncChannel>>;
 impl Finalize for JsAsyncChannel {}
 
 fn js_async_channel_resolve(mut cx: FunctionContext) -> JsResult<JsUndefined> {
-    #[cfg(build = "debug")]
-    trace!("JsAsyncChannel.resolved");
-
     let this = cx.this().downcast_or_throw::<BoxedChannel, _>(&mut cx)?;
+
+    #[cfg(debug_assertions)]
+    trace!("JsAsyncChannel.resolved {}", this.borrow()._id);
+
     let result = cx.argument::<JsValue>(0)?;
 
-    if this.borrow_mut().resolve(&mut cx, result) {
-        Ok(cx.undefined())
+    let tricky_rust_scope_hack = if let Err(err) = this.borrow_mut().resolve(&mut cx, result) {
+        cx.throw_error(format!("JsAsyncChannel resolving error: {}", err))
     } else {
-        cx.throw_error("Resolve was called on AsyncChannel that was already used")
-    }
+        Ok(cx.undefined())
+    };
+
+    tricky_rust_scope_hack
 }
 
 fn js_async_channel_reject(mut cx: FunctionContext) -> JsResult<JsUndefined> {
-    #[cfg(build = "debug")]
-    trace!("JsAsyncChannel.reject");
-
     let this = cx.this().downcast_or_throw::<BoxedChannel, _>(&mut cx)?;
+
+    #[cfg(debug_assertions)]
+    trace!("JsAsyncChannel.reject {}", this.borrow()._id);
+
     let error = cx.argument::<JsString>(0)?;
 
     let error_str = error.value(&mut cx);
-    if this.borrow_mut().reject(&mut cx, error_str) {
-        Ok(cx.undefined())
+
+    let tricky_rust_scope_hack = if let Err(err) = this.borrow_mut().reject(&mut cx, error_str) {
+        cx.throw_error(format!("JsAsyncChannel rejecting error: {}", err))
     } else {
-        cx.throw_error("Reject was called on AsyncChannel that was already used")
-    }
+        Ok(cx.undefined())
+    };
+
+    tricky_rust_scope_hack
 }
 
 impl JsAsyncChannel {
@@ -70,6 +87,8 @@ impl JsAsyncChannel {
 
     pub fn new_raw(callback: JsAsyncChannelCallback) -> Self {
         Self {
+            #[cfg(debug_assertions)]
+            _id: JS_ASYNC_CHANNEL_DEBUG_ID_SEQ.fetch_add(1, Ordering::SeqCst),
             callback: Some(callback),
         }
     }
@@ -91,23 +110,27 @@ impl JsAsyncChannel {
         Ok(obj)
     }
 
-    fn resolve(&mut self, cx: &mut FunctionContext, result: Handle<JsValue>) -> bool {
+    fn resolve(
+        &mut self,
+        cx: &mut FunctionContext,
+        result: Handle<JsValue>,
+    ) -> Result<(), CubeError> {
         if let Some(callback) = self.callback.take() {
-            callback(cx, Ok(result));
-
-            true
+            callback(cx, Ok(result))
         } else {
-            false
+            Err(CubeError::internal(
+                "Resolve was called on AsyncChannel that was already used".to_string(),
+            ))
         }
     }
 
-    fn reject(&mut self, cx: &mut FunctionContext, error: String) -> bool {
+    fn reject(&mut self, cx: &mut FunctionContext, error: String) -> Result<(), CubeError> {
         if let Some(callback) = self.callback.take() {
-            callback(cx, Err(CubeError::internal(error)));
-
-            true
+            callback(cx, Err(CubeError::internal(error)))
         } else {
-            false
+            Err(CubeError::internal(
+                "Reject was called on AsyncChannel that was already used".to_string(),
+            ))
         }
     }
 }
@@ -132,7 +155,12 @@ where
             Err(err) => Err(CubeError::internal(err.to_string())),
         };
 
-        tx.send(to_channel).unwrap();
+        tx.send(to_channel).map_err(|_| {
+            CubeError::internal(
+                "AsyncChannel: Unable to send result from JS back to Rust, channel closed"
+                    .to_string(),
+            )
+        })
     }));
 
     channel
@@ -185,7 +213,12 @@ where
     let async_channel = JsAsyncChannel::new_raw(Box::new(move |cx, result| {
         let to_channel = result.and_then(|res| result_from_js_value(cx, res));
 
-        tx.send(to_channel).unwrap();
+        tx.send(to_channel).map_err(|_| {
+            CubeError::internal(
+                "AsyncChannel: Unable to send result from JS back to Rust, channel closed"
+                    .to_string(),
+            )
+        })
     }));
 
     channel.send(move |mut cx| {