fix(native): Jinja - catch panic from render workers

packages/cubejs-backend-native/src/template/workers.rs

@@ -1,10 +1,11 @@
 use crate::template::neon_mj::*;
 use cubesql::CubeError;
 
-use log::trace;
+use log::{error, trace};
 use minijinja as mj;
 use neon::prelude::*;
 use neon::types::Deferred;
+use std::panic;
 
 pub struct JinjaEngineWorkerJob {
     pub(crate) template_name: String,
@@ -17,38 +18,67 @@ struct JinjaEngineWorker {
 }
 
 impl JinjaEngineWorker {
+    /// Renders a template, converting an unwinding panic into an error.
+    ///
+    /// Rendering executes user provided code (Python filters/functions via pyo3, custom
+    /// value implementations), which can panic. Without catching it, the worker thread
+    /// dies: the promise is rejected with a useless `Deferred` was dropped without being
+    /// settled error and the pool silently shrinks. Once the last worker is gone, the job
+    /// channel is closed and every next render fails with "sending into a closed channel".
+    fn render_catch_panic(
+        env: &mj::Environment,
+        template_name: &str,
+        ctx: mj::value::Value,
+    ) -> Result<Result<String, mj::Error>, CubeError> {
+        let render_block = panic::AssertUnwindSafe(|| {
+            let template = env.get_template(template_name)?;
+
+            template.render(ctx)
+        });
+
+        panic::catch_unwind(render_block).map_err(|panic_payload| {
+            CubeError::panic_with_message(
+                panic_payload,
+                "Unexpected panic while rendering jinja template",
+            )
+        })
+    }
+
+    fn process_render(job: JinjaEngineWorkerJob, js_channel: &Channel, env: &mj::Environment) {
+        let JinjaEngineWorkerJob {
+            template_name,
+            ctx,
+            deferred,
+        } = job;
+
+        match Self::render_catch_panic(env, &template_name, ctx) {
+            Ok(result) => {
+                deferred.settle_with(js_channel, move |mut cx| -> NeonResult<Handle<JsString>> {
+                    match result {
+                        Ok(r) => Ok(cx.string(r)),
+                        Err(err) => cx.throw_from_mj_error(err),
+                    }
+                });
+            }
+            Err(err) => {
+                error!("{} (template: {})", err, template_name);
+
+                deferred.settle_with(js_channel, move |mut cx| -> NeonResult<Handle<JsString>> {
+                    cx.throw_error(err.to_string())
+                });
+            }
+        }
+    }
+
     fn new(
         id: usize,
         env: mj::Environment<'static>,
-        js_channel: neon::event::Channel,
+        js_channel: Channel,
         receiver: async_channel::Receiver<JinjaEngineWorkerJob>,
     ) -> Self {
         let thread = std::thread::spawn(move || loop {
             if let Ok(job) = receiver.recv_blocking() {
-                let template = match env.get_template(&job.template_name) {
-                    Ok(t) => t,
-                    Err(err) => {
-                        job.deferred.settle_with(
-                            &js_channel,
-                            move |mut cx| -> NeonResult<Handle<JsString>> {
-                                cx.throw_from_mj_error(err)
-                            },
-                        );
-
-                        continue;
-                    }
-                };
-
-                let result = template.render(job.ctx);
-                job.deferred.settle_with(
-                    &js_channel,
-                    move |mut cx| -> NeonResult<Handle<JsString>> {
-                        match result {
-                            Ok(r) => Ok(cx.string(r)),
-                            Err(err) => cx.throw_from_mj_error(err),
-                        }
-                    },
-                );
+                Self::process_render(job, &js_channel, &env);
             } else {
                 trace!(
                     "Closing jinja thread, id: {}, threadId: {:?}",
@@ -101,3 +131,77 @@ impl JinjaEngineWorkerPool {
             .map_err(|err| CubeError::internal(format!("Unable to schedule rendering: {}", err)))
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn test_environment() -> mj::Environment<'static> {
+        let mut env = mj::Environment::new();
+
+        env.add_function("panic_fn", || -> Result<mj::value::Value, mj::Error> {
+            panic!("Boom from a function")
+        });
+
+        env.add_template("render.jinja", "Hello {{ name }}")
+            .unwrap();
+        env.add_template("panic.jinja", "{{ panic_fn() }}").unwrap();
+
+        env
+    }
+
+    #[test]
+    fn test_render() {
+        let env = test_environment();
+
+        let actual = JinjaEngineWorker::render_catch_panic(
+            &env,
+            "render.jinja",
+            mj::context! { name => "world" },
+        )
+        .expect("Render must not panic")
+        .expect("Render must not fail");
+
+        assert_eq!(actual, "Hello world");
+    }
+
+    #[test]
+    fn test_render_unknown_template() {
+        let env = test_environment();
+
+        let err = JinjaEngineWorker::render_catch_panic(
+            &env,
+            "unknown.jinja",
+            mj::value::Value::UNDEFINED,
+        )
+        .expect("Unknown template must not panic")
+        .expect_err("Unknown template must fail");
+
+        assert_eq!(err.kind(), mj::ErrorKind::TemplateNotFound);
+    }
+
+    #[test]
+    fn test_render_panic() {
+        let env = test_environment();
+
+        let err =
+            JinjaEngineWorker::render_catch_panic(&env, "panic.jinja", mj::value::Value::UNDEFINED)
+                .expect_err("Panic must be caught");
+
+        assert_eq!(
+            err.message,
+            "Unexpected panic while rendering jinja template. Reason: Boom from a function"
+        );
+
+        // Environment must stay usable after a panic, because the worker is reused
+        let actual = JinjaEngineWorker::render_catch_panic(
+            &env,
+            "render.jinja",
+            mj::context! { name => "world" },
+        )
+        .expect("Render must not panic")
+        .expect("Render must not fail");
+
+        assert_eq!(actual, "Hello world");
+    }
+}

packages/cubejs-backend-native/test/jinja.test.ts

@@ -157,6 +157,7 @@ suite('Jinja (new api)', () => {
     loadTemplateFile(jinjaEngine, 'variables.yml.jinja');
     loadTemplateFile(jinjaEngine, 'filters.yml.jinja');
     loadTemplateFile(jinjaEngine, 'template_error_python.jinja');
+    loadTemplateFile(jinjaEngine, 'template_panic.jinja');
 
     for (let i = 1; i < 9; i++) {
       loadTemplateFile(jinjaEngine, `0${i}.yml.jinja`);
@@ -192,4 +193,19 @@ suite('Jinja (new api)', () => {
   for (let i = 1; i < 9; i++) {
     testTemplateBySnapshot(initJinjaEngine, `0${i}.yml.jinja`, {});
   }
+
+  // A panic inside a worker thread used to kill it, which left the promise
+  // pending forever and reduced the pool by one worker
+  test('render template_panic.jinja', async () => {
+    const { jinjaEngine } = await initJinjaEngine();
+
+    await expect(
+      jinjaEngine.renderTemplate('template_panic.jinja', { js_fn: () => 'unsupported' }, null)
+    ).rejects.toThrow(/Unexpected panic while rendering jinja template/);
+
+    // The worker must survive a panic and continue to process the next jobs
+    await expect(
+      jinjaEngine.renderTemplate('01.yml.jinja', {}, null)
+    ).resolves.toBeDefined();
+  });
 });

packages/cubejs-backend-native/test/templates/template_panic.jinja

@@ -0,0 +1,3 @@
+{# Conversion of a JsFunction to a minijinja value is not supported and panics,
+   it's used to test that a panic inside a worker doesn't kill it #}
+{{ COMPILE_CONTEXT.js_fn }}