From 10efc7124490d54187ca678124105ab0cea1add6 Mon Sep 17 00:00:00 2001 From: Stefan Date: Mon, 20 Jul 2026 11:43:22 +0200 Subject: [PATCH] feat: implement new loaders across languages --- Cargo.lock | 1 + bindings/c/Cargo.toml | 1 + bindings/c/build.rs | 11 +- bindings/c/src/engine.rs | 270 +++++++++++++++++++++++++++++++- bindings/c/src/error.rs | 5 + bindings/c/src/languages/go.rs | 24 ++- bindings/c/src/loader.rs | 59 ++++++- bindings/c/src/mt.rs | 9 ++ bindings/c/src/result.rs | 11 ++ bindings/c/zen_engine.h | 61 +++++++- bindings/python/src/engine.rs | 139 ++++++++++++++-- bindings/python/test_sync.py | 56 +++++++ bindings/python/zen.pyi | 42 ++++- bindings/uniffi/Cargo.toml | 3 + bindings/uniffi/nuget/README.md | 12 ++ bindings/uniffi/src/engine.rs | 208 ++++++++++++++++++++++-- bindings/uniffi/src/loader.rs | 54 ++++++- bindings/uniffi/src/types.rs | 13 ++ core/engine/src/engine.rs | 2 +- 19 files changed, 938 insertions(+), 43 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index c11787ae..67edabf3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5642,6 +5642,7 @@ dependencies = [ "serde_json", "strum", "tokio", + "tokio-util", "zen-engine", "zen-expression", "zen-tmpl", diff --git a/bindings/c/Cargo.toml b/bindings/c/Cargo.toml index fc6209b1..3180829e 100644 --- a/bindings/c/Cargo.toml +++ b/bindings/c/Cargo.toml @@ -12,6 +12,7 @@ serde = { workspace = true } serde_json = { workspace = true } strum = { workspace = true, features = ["derive"] } tokio = { workspace = true, features = ["rt"] } +tokio-util = { workspace = true, features = ["rt"] } zen-engine = { path = "../../core/engine", features = ["arbitrary_precision"] } zen-expression = { path = "../../core/expression", features = ["arbitrary_precision"] } zen-tmpl = { path = "../../core/template" } diff --git a/bindings/c/build.rs b/bindings/c/build.rs index 01af963a..a884235f 100644 --- a/bindings/c/build.rs +++ b/bindings/c/build.rs @@ -3,8 +3,17 @@ use std::env; fn main() { let crate_dir = env::var("CARGO_MANIFEST_DIR").unwrap(); + let config = cbindgen::Config { + language: cbindgen::Language::C, + enumeration: cbindgen::EnumConfig { + prefix_with_name: true, + ..Default::default() + }, + ..Default::default() + }; + cbindgen::Builder::new() - .with_language(cbindgen::Language::C) + .with_config(config) .with_crate(crate_dir) .generate() .expect("Unable to generate bindings") diff --git a/bindings/c/src/engine.rs b/bindings/c/src/engine.rs index 9a2b6340..42c49fdd 100644 --- a/bindings/c/src/engine.rs +++ b/bindings/c/src/engine.rs @@ -6,12 +6,15 @@ use std::sync::Arc; use zen_engine::{DecisionEngine, EvaluationOptions}; use crate::custom_node::DynamicCustomNode; +use crate::custom_node::ZenCustomNodeResult; use crate::decision::{ZenDecision, ZenDecisionStruct}; use crate::error::ZenError; use crate::helper::safe_str_from_ptr; -use crate::loader::DynamicDecisionLoader; -use crate::mt::tokio_runtime; +use crate::languages::native::NativeCustomNode; +use crate::loader::{DynamicDecisionLoader, ZenEngineLoaderConfig}; +use crate::mt::{tokio_runtime, worker_pool}; use crate::result::ZenResult; +use serde_json::json; pub(crate) struct ZenEngine(DecisionEngine); @@ -72,6 +75,29 @@ pub extern "C" fn zen_engine_new() -> *mut ZenEngineStruct { Box::into_raw(Box::new(ZenEngine::default())) as *mut ZenEngineStruct } +/// Creates a new ZenEngine instance from a loader configuration, caller is responsible for +/// freeing the returned reference by calling zen_engine_free. +#[no_mangle] +pub extern "C" fn zen_engine_new_with_loader_config( + config: ZenEngineLoaderConfig, + maybe_custom_node: Option ZenCustomNodeResult>, +) -> ZenResult { + let loader = match config.to_dynamic_loader() { + Ok(loader) => loader, + Err(error) => return ZenResult::error(error), + }; + + let custom_node = match maybe_custom_node { + Some(callback) => DynamicCustomNode::Native(NativeCustomNode::new(callback)), + None => DynamicCustomNode::default(), + }; + + let engine = ZenEngine::new(DynamicDecisionLoader::Config(loader), custom_node); + engine.compile(); + + ZenResult::ok(Box::into_raw(Box::new(engine)) as *mut ZenEngineStruct) +} + /// Frees the ZenEngine instance reference from the memory #[no_mangle] pub extern "C" fn zen_engine_free(engine: *mut ZenEngineStruct) { @@ -148,6 +174,102 @@ pub extern "C" fn zen_engine_evaluate( ZenResult::ok(cstring_result.into_raw()) } +#[repr(C)] +pub struct ZenEngineEvaluateBatchRequest { + key: *const c_char, + context: *const c_char, +} + +enum BatchTask { + Failed(serde_json::Value), + Pending(tokio::task::JoinHandle>), +} + +/// Evaluates a batch of requests in parallel using a DecisionEngine reference via loader. +/// Returns a JSON array of { success, data?, error? } envelopes in request order. +/// Caller is responsible for freeing: requests and ZenResult. +#[no_mangle] +pub extern "C" fn zen_engine_evaluate_batch( + engine: *const ZenEngineStruct, + requests: *const ZenEngineEvaluateBatchRequest, + requests_len: usize, + options: ZenEngineEvaluationOptions, +) -> ZenResult { + if engine.is_null() || (requests.is_null() && requests_len > 0) { + return ZenResult::error(ZenError::InvalidArgument); + } + + let request_slice = match requests_len { + 0 => &[], + _ => unsafe { std::slice::from_raw_parts(requests, requests_len) }, + }; + + let mut parsed: Vec<(String, Result)> = Vec::with_capacity(requests_len); + for request in request_slice { + let Some(key) = safe_str_from_ptr(request.key) else { + return ZenResult::error(ZenError::InvalidArgument); + }; + + if request.context.is_null() { + return ZenResult::error(ZenError::InvalidArgument); + } + + let cstr_context = unsafe { CStr::from_ptr(request.context) }; + let context = + serde_json::from_slice::(cstr_context.to_bytes()).map_err(|e| e.to_string()); + parsed.push((key.to_string(), context)); + } + + let zen_engine = unsafe { &*(engine as *const ZenEngine) }; + let decision_engine: DecisionEngine = DecisionEngine::clone(zen_engine); + let eval_options: EvaluationOptions = options.into(); + + let pool = worker_pool(); + let tasks: Vec = parsed + .into_iter() + .map(|(key, context)| match context { + Err(message) => BatchTask::Failed(json!(format!("invalid context: {message}"))), + Ok(value) => { + let engine = decision_engine.clone(); + BatchTask::Pending(pool.spawn_pinned(move || async move { + engine + .evaluate_with_opts(key, value.into(), eval_options) + .await + .map(|response| serde_json::to_value(&response).unwrap_or(Value::Null)) + .map_err(|e| { + serde_json::to_value(&e).unwrap_or_else(|_| json!(e.to_string())) + }) + })) + } + }) + .collect(); + + let results = tokio_runtime().block_on(async move { + let mut items = Vec::with_capacity(tasks.len()); + for task in tasks { + let item = match task { + BatchTask::Failed(error) => json!({ "success": false, "error": error }), + BatchTask::Pending(handle) => match handle.await { + Ok(Ok(data)) => json!({ "success": true, "data": data }), + Ok(Err(error)) => json!({ "success": false, "error": error }), + Err(_) => { + json!({ "success": false, "error": "evaluation worker panicked" }) + } + }, + }; + items.push(item); + } + Value::Array(items) + }); + + let Ok(serialized_results) = serde_json::to_string(&results) else { + return ZenResult::error(ZenError::JsonSerializationFailed); + }; + + let cstring_result = unsafe { CString::from_vec_unchecked(serialized_results.into_bytes()) }; + ZenResult::ok(cstring_result.into_raw()) +} + /// Loads a Decision through DecisionEngine /// Caller is responsible for freeing: key and ZenResult. #[no_mangle] @@ -174,3 +296,147 @@ pub extern "C" fn zen_engine_get_decision( let zen_decision = ZenDecision::from(decision); ZenResult::ok(Box::into_raw(Box::new(zen_decision)) as *mut ZenDecisionStruct) } + +#[cfg(test)] +mod tests { + use super::*; + use crate::error::ZenErrorDiscriminants; + use crate::loader::ZenLoaderConfigKind; + use std::ffi::CString; + use std::ptr::null; + + fn evaluate_table(engine: *mut ZenEngineStruct) -> Value { + let key = CString::new("table.json").unwrap(); + let context = CString::new(r#"{"input":12}"#).unwrap(); + let result = zen_engine_evaluate( + engine, + key.as_ptr(), + context.as_ptr(), + ZenEngineEvaluationOptions { + trace: false, + max_depth: 5, + }, + ); + + assert_eq!(result.error_code(), 0); + let response = unsafe { CString::from_raw(result.result_ptr()) }; + serde_json::from_slice(response.to_bytes()).unwrap() + } + + #[test] + fn engine_from_static_loader_config() { + let content = CString::new(format!( + r#"{{"table.json": {}}}"#, + include_str!("../../../test-data/table.json") + )) + .unwrap(); + + let config = ZenEngineLoaderConfig { + kind: ZenLoaderConfigKind::Static, + content: content.as_ptr(), + bytes: null(), + bytes_len: 0, + }; + + let result = zen_engine_new_with_loader_config(config, None); + assert_eq!(result.error_code(), 0); + + let engine = result.result_ptr(); + let response = evaluate_table(engine); + assert_eq!(response["result"]["output"], serde_json::json!(10)); + + zen_engine_free(engine); + } + + #[test] + fn engine_from_fs_loader_config() { + let path = CString::new(concat!(env!("CARGO_MANIFEST_DIR"), "/../../test-data")).unwrap(); + + let config = ZenEngineLoaderConfig { + kind: ZenLoaderConfigKind::Filesystem, + content: path.as_ptr(), + bytes: null(), + bytes_len: 0, + }; + + let result = zen_engine_new_with_loader_config(config, None); + assert_eq!(result.error_code(), 0); + + let engine = result.result_ptr(); + let response = evaluate_table(engine); + assert_eq!(response["result"]["output"], serde_json::json!(10)); + + zen_engine_free(engine); + } + + #[test] + fn engine_evaluate_batch_mixed_results() { + let path = CString::new(concat!(env!("CARGO_MANIFEST_DIR"), "/../../test-data")).unwrap(); + let config = ZenEngineLoaderConfig { + kind: ZenLoaderConfigKind::Filesystem, + content: path.as_ptr(), + bytes: null(), + bytes_len: 0, + }; + let engine = zen_engine_new_with_loader_config(config, None).result_ptr(); + + let keys = [ + CString::new("table.json").unwrap(), + CString::new("missing.json").unwrap(), + CString::new("table.json").unwrap(), + ]; + let contexts = [ + CString::new(r#"{"input":12}"#).unwrap(), + CString::new(r#"{}"#).unwrap(), + CString::new(r#"{"input":5}"#).unwrap(), + ]; + let requests: Vec = keys + .iter() + .zip(contexts.iter()) + .map(|(key, context)| ZenEngineEvaluateBatchRequest { + key: key.as_ptr(), + context: context.as_ptr(), + }) + .collect(); + + let result = zen_engine_evaluate_batch( + engine, + requests.as_ptr(), + requests.len(), + ZenEngineEvaluationOptions { + trace: false, + max_depth: 5, + }, + ); + + assert_eq!(result.error_code(), 0); + let response = unsafe { CString::from_raw(result.result_ptr()) }; + let items: Value = serde_json::from_slice(response.to_bytes()).unwrap(); + + assert_eq!(items[0]["success"], serde_json::json!(true)); + assert_eq!(items[0]["data"]["result"]["output"], serde_json::json!(10)); + assert_eq!(items[1]["success"], serde_json::json!(false)); + assert_eq!(items[2]["success"], serde_json::json!(true)); + assert_eq!(items[2]["data"]["result"]["output"], serde_json::json!(0)); + + zen_engine_free(engine); + } + + #[test] + fn engine_from_invalid_zip_loader_config() { + let bytes = [0u8; 4]; + let config = ZenEngineLoaderConfig { + kind: ZenLoaderConfigKind::Zip, + content: null(), + bytes: bytes.as_ptr(), + bytes_len: bytes.len(), + }; + + let result = zen_engine_new_with_loader_config(config, None); + assert_eq!( + result.error_code(), + ZenErrorDiscriminants::LoaderConfigError as u8 + ); + assert!(result.result_ptr().is_null()); + } +} diff --git a/bindings/c/src/error.rs b/bindings/c/src/error.rs index 2345340a..a62dde5f 100644 --- a/bindings/c/src/error.rs +++ b/bindings/c/src/error.rs @@ -21,6 +21,8 @@ pub enum ZenError { LoaderInternalError { key: String, message: String }, TemplateEngineError { template: String, message: String }, + + LoaderConfigError { message: String }, } impl ZenError { @@ -35,6 +37,9 @@ impl ZenError { ZenError::TemplateEngineError { template, message } => { Some(json!({ "template": template, "message": message }).to_string()) } + ZenError::LoaderConfigError { message } => { + Some(json!({ "message": message }).to_string()) + } _ => None, } } diff --git a/bindings/c/src/languages/go.rs b/bindings/c/src/languages/go.rs index 12ee9c33..1aefd66d 100644 --- a/bindings/c/src/languages/go.rs +++ b/bindings/c/src/languages/go.rs @@ -1,6 +1,7 @@ use crate::custom_node::{DynamicCustomNode, ZenCustomNodeResult}; use crate::engine::{ZenEngine, ZenEngineStruct}; -use crate::loader::{DynamicDecisionLoader, ZenDecisionLoaderResult}; +use crate::loader::{DynamicDecisionLoader, ZenDecisionLoaderResult, ZenEngineLoaderConfig}; +use crate::result::ZenResult; use std::ffi::{c_char, CString}; use std::future::Future; use std::pin::Pin; @@ -103,6 +104,27 @@ pub extern "C" fn zen_engine_new_golang( Box::into_raw(Box::new(engine)) as *mut ZenEngineStruct } +/// Creates a DecisionEngine from a loader configuration using GoLang handler (optional). Caller is responsible for freeing DecisionEngine. +#[no_mangle] +pub extern "C" fn zen_engine_new_golang_with_loader_config( + config: ZenEngineLoaderConfig, + maybe_custom_node: Option<&usize>, +) -> ZenResult { + let loader = match config.to_dynamic_loader() { + Ok(loader) => loader, + Err(error) => return ZenResult::error(error), + }; + + let custom_node = GoCustomNode::new(map_handler(maybe_custom_node.cloned())); + let engine = ZenEngine::new( + DynamicDecisionLoader::Config(loader), + DynamicCustomNode::Go(custom_node), + ); + engine.compile(); + + ZenResult::ok(Box::into_raw(Box::new(engine)) as *mut ZenEngineStruct) +} + #[allow(unused_doc_comments)] /// cbindgen:ignore extern "C" { diff --git a/bindings/c/src/loader.rs b/bindings/c/src/loader.rs index ab7e0711..ecc457b0 100644 --- a/bindings/c/src/loader.rs +++ b/bindings/c/src/loader.rs @@ -5,15 +5,20 @@ use std::sync::Arc; use anyhow::anyhow; -use zen_engine::loader::{DecisionLoader, LoaderError, LoaderResponse, NoopLoader}; +use zen_engine::loader::{ + DecisionLoader, DynamicLoader, LoaderConfig, LoaderError, LoaderResponse, NoopLoader, +}; use zen_engine::model::DecisionContent; +use crate::error::ZenError; +use crate::helper::safe_str_from_ptr; use crate::languages::native::NativeDecisionLoader; #[derive(Debug)] pub(crate) enum DynamicDecisionLoader { Noop(NoopLoader), Native(NativeDecisionLoader), + Config(DynamicLoader), #[cfg(feature = "go")] Go(crate::languages::go::GoDecisionLoader), } @@ -33,6 +38,7 @@ impl DecisionLoader for DynamicDecisionLoader { match self { DynamicDecisionLoader::Noop(loader) => loader.load(key).await, DynamicDecisionLoader::Native(loader) => loader.load(key).await, + DynamicDecisionLoader::Config(loader) => loader.load(key).await, #[cfg(feature = "go")] DynamicDecisionLoader::Go(loader) => loader.load(key).await, } @@ -40,6 +46,57 @@ impl DecisionLoader for DynamicDecisionLoader { } } +#[allow(dead_code)] +#[repr(C)] +pub enum ZenLoaderConfigKind { + Static, + Filesystem, + Zip, +} + +#[repr(C)] +pub struct ZenEngineLoaderConfig { + pub(crate) kind: ZenLoaderConfigKind, + pub(crate) content: *const c_char, + pub(crate) bytes: *const u8, + pub(crate) bytes_len: usize, +} + +impl ZenEngineLoaderConfig { + pub(crate) fn to_dynamic_loader(&self) -> Result { + let config = match self.kind { + ZenLoaderConfigKind::Static => { + let content = safe_str_from_ptr(self.content).ok_or(ZenError::InvalidArgument)?; + LoaderConfig::Static { + content: serde_json::from_str(content) + .map_err(|_| ZenError::JsonDeserializationFailed)?, + } + } + ZenLoaderConfigKind::Filesystem => { + let path = safe_str_from_ptr(self.content).ok_or(ZenError::InvalidArgument)?; + LoaderConfig::Filesystem { + path: path.to_string(), + } + } + ZenLoaderConfigKind::Zip => { + if self.bytes.is_null() { + return Err(ZenError::InvalidArgument); + } + + let bytes = + unsafe { std::slice::from_raw_parts(self.bytes, self.bytes_len) }.to_vec(); + LoaderConfig::Zip { bytes } + } + }; + + config + .into_loader() + .map_err(|e| ZenError::LoaderConfigError { + message: e.to_string(), + }) + } +} + #[repr(C)] pub struct ZenDecisionLoaderResult { content: *mut c_char, diff --git a/bindings/c/src/mt.rs b/bindings/c/src/mt.rs index 950607cd..74f19ffa 100644 --- a/bindings/c/src/mt.rs +++ b/bindings/c/src/mt.rs @@ -1,7 +1,16 @@ use std::sync::{Arc, OnceLock}; +use std::thread::available_parallelism; use tokio::runtime; use tokio::runtime::Runtime; +use tokio_util::task::LocalPoolHandle; + +pub(crate) fn worker_pool() -> LocalPoolHandle { + static LOCAL_POOL: OnceLock = OnceLock::new(); + LOCAL_POOL + .get_or_init(|| LocalPoolHandle::new(available_parallelism().map(Into::into).unwrap_or(1))) + .clone() +} pub(crate) fn tokio_runtime() -> Arc { static RUNTIME: OnceLock> = OnceLock::new(); diff --git a/bindings/c/src/result.rs b/bindings/c/src/result.rs index 47644f80..117ad45d 100644 --- a/bindings/c/src/result.rs +++ b/bindings/c/src/result.rs @@ -51,6 +51,17 @@ impl ZenResult { } } +#[cfg(test)] +impl ZenResult { + pub(crate) fn result_ptr(&self) -> *mut T { + self.result + } + + pub(crate) fn error_code(&self) -> u8 { + self.error + } +} + impl From<&Box> for ZenResult { fn from(evaluation_error: &Box) -> Self { let Ok(value) = serde_json::to_value(evaluation_error) else { diff --git a/bindings/c/zen_engine.h b/bindings/c/zen_engine.h index fb94c00b..c76ecc1a 100644 --- a/bindings/c/zen_engine.h +++ b/bindings/c/zen_engine.h @@ -3,6 +3,12 @@ #include #include +typedef enum ZenLoaderConfigKind { + ZenLoaderConfigKind_Static, + ZenLoaderConfigKind_Filesystem, + ZenLoaderConfigKind_Zip, +} ZenLoaderConfigKind; + typedef struct ZenDecisionStruct { uint8_t _data[0]; } ZenDecisionStruct; @@ -26,6 +32,28 @@ typedef struct ZenEngineStruct { uint8_t _data[0]; } ZenEngineStruct; +/** + * CResult can be seen as Either. It cannot, and should not, be initialized + * manually. Instead, use error or ok functions for initialisation. + */ +typedef struct ZenResult_ZenEngineStruct { + struct ZenEngineStruct *result; + uint8_t error; + char *details; +} ZenResult_ZenEngineStruct; + +typedef struct ZenEngineLoaderConfig { + enum ZenLoaderConfigKind kind; + const char *content; + const uint8_t *bytes; + uintptr_t bytes_len; +} ZenEngineLoaderConfig; + +typedef struct ZenCustomNodeResult { + char *content; + char *error; +} ZenCustomNodeResult; + /** * CResult can be seen as Either. It cannot, and should not, be initialized * manually. Instead, use error or ok functions for initialisation. @@ -36,6 +64,11 @@ typedef struct ZenResult_ZenDecisionStruct { char *details; } ZenResult_ZenDecisionStruct; +typedef struct ZenEngineEvaluateBatchRequest { + const char *key; + const char *context; +} ZenEngineEvaluateBatchRequest; + /** * CResult can be seen as Either. It cannot, and should not, be initialized * manually. Instead, use error or ok functions for initialisation. @@ -53,11 +86,6 @@ typedef struct ZenDecisionLoaderResult { typedef struct ZenDecisionLoaderResult (*ZenDecisionLoaderNativeCallback)(const char *key); -typedef struct ZenCustomNodeResult { - char *content; - char *error; -} ZenCustomNodeResult; - typedef struct ZenCustomNodeResult (*ZenCustomNodeNativeCallback)(const char *request); /** @@ -79,6 +107,13 @@ struct ZenResult_c_char zen_decision_evaluate(const struct ZenDecisionStruct *de */ struct ZenEngineStruct *zen_engine_new(void); +/** + * Creates a new ZenEngine instance from a loader configuration, caller is responsible for + * freeing the returned reference by calling zen_engine_free. + */ +struct ZenResult_ZenEngineStruct zen_engine_new_with_loader_config(struct ZenEngineLoaderConfig config, + struct ZenCustomNodeResult (*maybe_custom_node)(const char *request)); + /** * Frees the ZenEngine instance reference from the memory */ @@ -100,6 +135,16 @@ struct ZenResult_c_char zen_engine_evaluate(const struct ZenEngineStruct *engine const char *context, struct ZenEngineEvaluationOptions options); +/** + * Evaluates a batch of requests in parallel using a DecisionEngine reference via loader. + * Returns a JSON array of { success, data?, error? } envelopes in request order. + * Caller is responsible for freeing: requests and ZenResult. + */ +struct ZenResult_c_char zen_engine_evaluate_batch(const struct ZenEngineStruct *engine, + const struct ZenEngineEvaluateBatchRequest *requests, + uintptr_t requests_len, + struct ZenEngineEvaluationOptions options); + /** * Loads a Decision through DecisionEngine * Caller is responsible for freeing: key and ZenResult. @@ -135,3 +180,9 @@ struct ZenEngineStruct *zen_engine_new_native(ZenDecisionLoaderNativeCallback lo */ struct ZenEngineStruct *zen_engine_new_golang(const uintptr_t *maybe_loader, const uintptr_t *maybe_custom_node); + +/** + * Creates a DecisionEngine from a loader configuration using GoLang handler (optional). Caller is responsible for freeing DecisionEngine. + */ +struct ZenResult_ZenEngineStruct zen_engine_new_golang_with_loader_config(struct ZenEngineLoaderConfig config, + const uintptr_t *maybe_custom_node); diff --git a/bindings/python/src/engine.rs b/bindings/python/src/engine.rs index 07f116ba..473d0b03 100644 --- a/bindings/python/src/engine.rs +++ b/bindings/python/src/engine.rs @@ -8,13 +8,15 @@ use crate::mt::{block_on, worker_pool}; use crate::value::PyValue; use crate::variable::PyVariable; use anyhow::{anyhow, Context}; -use pyo3::prelude::{PyAnyMethods, PyDictMethods}; -use pyo3::types::PyDict; +use pyo3::prelude::{PyAnyMethods, PyDictMethods, PyListMethods}; +use pyo3::types::{PyDict, PyList}; use pyo3::{pyclass, pymethods, Bound, FromPyObject, IntoPyObjectExt, Py, PyAny, PyResult, Python}; use pyo3_async_runtimes::tokio::get_current_locals; use pyo3_async_runtimes::{tokio, TaskLocals}; +use pythonize::{depythonize, pythonize}; use serde::{Deserialize, Serialize}; use serde_json::Value; +use zen_engine::loader::{DynamicLoader, LoaderConfig}; use zen_engine::{DecisionEngine, EvaluationOptions}; #[pyclass] @@ -76,6 +78,53 @@ impl Default for PyZenEngine { } } +pub struct PyBatchRequest { + key: String, + context: Value, +} + +impl<'py> FromPyObject<'py> for PyBatchRequest { + fn extract_bound(ob: &Bound<'py, PyAny>) -> PyResult { + let dict = ob.downcast::()?; + + let key: String = dict + .get_item("key")? + .ok_or_else(|| anyhow!("batch request requires a 'key'"))? + .extract()?; + + let context = dict + .get_item("context")? + .ok_or_else(|| anyhow!("batch request requires a 'context'"))? + .extract::()?; + + Ok(Self { + key, + context: context.0, + }) + } +} + +impl PyZenEngine { + fn config_loader(config: &Bound<'_, PyDict>) -> PyResult { + let loader_type: Option = + config.get_item("type")?.map(|v| v.extract()).transpose()?; + + let loader_config = match loader_type.as_deref() { + Some("zip") => { + let bytes = config + .get_item("bytes")? + .ok_or_else(|| anyhow!("zip loader requires a 'bytes' value"))?; + LoaderConfig::Zip { + bytes: bytes.extract()?, + } + } + _ => depythonize(config.as_any())?, + }; + + Ok(loader_config.into_loader()?) + } +} + #[pymethods] impl PyZenEngine { #[new] @@ -85,11 +134,6 @@ impl PyZenEngine { return Ok(Default::default()); }; - let loader = match options.get_item("loader")? { - Some(loader) => Some(loader.into_py_any(py)?), - None => None, - }; - let custom_node = match options.get_item("customHandler")? { Some(custom_node) => Some(custom_node.into_py_any(py)?), None => None, @@ -102,11 +146,25 @@ impl PyZenEngine { .flatten() }; + let loader: DynamicLoader = match options.get_item("loader")? { + Some(loader) => match loader.downcast::() { + Ok(config) => Self::config_loader(config)?, + Err(_) => Arc::new(PyDecisionLoader::new( + Some(loader.into_py_any(py)?), + make_locals(), + )), + }, + None => Arc::new(PyDecisionLoader::default()), + }; + + let engine = DecisionEngine::new( + loader, + Arc::new(PyCustomNode::new(custom_node, make_locals())), + ); + engine.compile(); + Ok(Self { - engine: Arc::new(DecisionEngine::new( - Arc::new(PyDecisionLoader::new(loader, make_locals())), - Arc::new(PyCustomNode::new(custom_node, make_locals())), - )), + engine: Arc::new(engine), }) } @@ -131,6 +189,65 @@ impl PyZenEngine { crate::convert::response_to_py(py, result) } + #[pyo3(signature = (requests, opts=None))] + pub fn evaluate_batch( + &self, + py: Python, + requests: Vec, + opts: Option, + ) -> PyResult> { + let options: EvaluationOptions = opts.unwrap_or_default().into(); + + let handles: Vec<_> = requests + .into_iter() + .map(|request| { + let engine = self.engine.clone(); + worker_pool().spawn_pinned(move || async move { + engine + .evaluate_with_opts(request.key, request.context.into(), options) + .await + .map(crate::convert::PortableResponse::build) + .map_err(|e| { + serde_json::to_value(e.as_ref()) + .unwrap_or_else(|_| Value::String(e.to_string())) + }) + }) + }) + .collect(); + + let results = py.allow_threads(|| { + block_on(async move { + let mut out = Vec::with_capacity(handles.len()); + for handle in handles { + out.push(handle.await); + } + out + }) + }); + + let list = PyList::empty(py); + for result in results { + let item = PyDict::new(py); + match result { + Ok(Ok(response)) => { + item.set_item("success", true)?; + item.set_item("data", response.into_py(py)?)?; + } + Ok(Err(error)) => { + item.set_item("success", false)?; + item.set_item("error", pythonize(py, &error)?)?; + } + Err(_) => { + item.set_item("success", false)?; + item.set_item("error", "evaluation worker panicked")?; + } + } + list.append(item)?; + } + + Ok(list.into_py_any(py)?) + } + #[pyo3(signature = (key, ctx, opts=None))] pub fn async_evaluate<'py>( &'py self, diff --git a/bindings/python/test_sync.py b/bindings/python/test_sync.py index 8472d6df..4d0acc0c 100644 --- a/bindings/python/test_sync.py +++ b/bindings/python/test_sync.py @@ -68,6 +68,62 @@ def test_engine_custom_handler(self): self.assertEqual(r2["result"]["sum"], 30) self.assertEqual(r3["result"]["sum"], 40) + def test_static_loader_config(self): + with open("../../test-data/table.json", "r") as f: + table_content = json.loads(f.read()) + + engine = zen.ZenEngine({"loader": {"type": "static", "content": {"table.json": table_content}}}) + r1 = engine.evaluate("table.json", {"input": 2}) + r2 = engine.evaluate("table.json", {"input": 12}) + + self.assertEqual(r1["result"]["output"], 0) + self.assertEqual(r2["result"]["output"], 10) + self.assertRaises(RuntimeError, engine.evaluate, "missing.json", {}) + + def test_fs_loader_config(self): + engine = zen.ZenEngine({"loader": {"type": "fs", "path": "../../test-data"}}) + r1 = engine.evaluate("table.json", {"input": 2}) + r2 = engine.evaluate("table.json", {"input": 12}) + + self.assertEqual(r1["result"]["output"], 0) + self.assertEqual(r2["result"]["output"], 10) + + def test_zip_loader_config(self): + import io + import zipfile + + buffer = io.BytesIO() + with zipfile.ZipFile(buffer, "w", zipfile.ZIP_DEFLATED) as archive: + with open("../../test-data/table.json", "rb") as f: + archive.writestr("table.json", f.read()) + + engine = zen.ZenEngine({"loader": {"type": "zip", "bytes": buffer.getvalue()}}) + r1 = engine.evaluate("table.json", {"input": 2}) + r2 = engine.evaluate("table.json", {"input": 12}) + + self.assertEqual(r1["result"]["output"], 0) + self.assertEqual(r2["result"]["output"], 10) + + def test_evaluate_batch(self): + engine = zen.ZenEngine({"loader": {"type": "fs", "path": "../../test-data"}}) + results = engine.evaluate_batch([ + {"key": "table.json", "context": {"input": 12}}, + {"key": "missing.json", "context": {}}, + {"key": "table.json", "context": {"input": 5}}, + ]) + + self.assertEqual(len(results), 3) + self.assertTrue(results[0]["success"]) + self.assertEqual(results[0]["data"]["result"]["output"], 10) + self.assertFalse(results[1]["success"]) + self.assertIn("error", results[1]) + self.assertTrue(results[2]["success"]) + self.assertEqual(results[2]["data"]["result"]["output"], 0) + + def test_evaluate_batch_empty(self): + engine = zen.ZenEngine({"loader": loader}) + self.assertEqual(engine.evaluate_batch([]), []) + def test_evaluate_expression(self): result = zen.evaluate_expression("sum(a)", {"a": [1, 2, 3, 4]}) self.assertEqual(result, 10) diff --git a/bindings/python/zen.pyi b/bindings/python/zen.pyi index ecf448a2..0da522b3 100644 --- a/bindings/python/zen.pyi +++ b/bindings/python/zen.pyi @@ -1,4 +1,4 @@ -from collections.abc import Awaitable +from collections.abc import Awaitable, Callable from typing import Any, Optional, TypedDict, Literal, TypeAlias, Union @@ -17,12 +17,50 @@ ZenContext: TypeAlias = Union[str, bytes, dict] ZenDecisionContentInput: TypeAlias = Union[str, ZenDecisionContent] +class StaticLoaderConfig(TypedDict): + type: Literal["static"] + content: dict[str, dict] + + +class FilesystemLoaderConfig(TypedDict): + type: Literal["fs"] + path: str + + +class ZipLoaderConfig(TypedDict): + type: Literal["zip"] + bytes: bytes + + +ZenLoaderConfig: TypeAlias = Union[StaticLoaderConfig, FilesystemLoaderConfig, ZipLoaderConfig] +ZenLoaderCallback: TypeAlias = Callable[[str], Union[str, dict, ZenDecisionContent, Awaitable[Union[str, dict, ZenDecisionContent]]]] + + +class ZenEngineOptions(TypedDict, total=False): + loader: Union[ZenLoaderCallback, ZenLoaderConfig] + customHandler: Callable + + +class EvaluateBatchRequest(TypedDict): + key: str + context: Any + + +class EvaluateBatchResult(TypedDict, total=False): + success: bool + data: EvaluateResponse + error: Any + + class ZenEngine: - def __init__(self, options: Optional[dict] = None) -> None: ... + def __init__(self, options: Optional[ZenEngineOptions] = None) -> None: ... def evaluate(self, key: str, context: ZenContext, options: Optional[DecisionEvaluateOptions] = None) -> EvaluateResponse: ... + def evaluate_batch(self, requests: list[EvaluateBatchRequest], + options: Optional[DecisionEvaluateOptions] = None) -> list[EvaluateBatchResult]: ... + def async_evaluate(self, key: str, context: ZenContext, options: Optional[DecisionEvaluateOptions] = None) -> \ Awaitable[EvaluateResponse]: ... diff --git a/bindings/uniffi/Cargo.toml b/bindings/uniffi/Cargo.toml index 56690aa3..5ecdc503 100644 --- a/bindings/uniffi/Cargo.toml +++ b/bindings/uniffi/Cargo.toml @@ -21,6 +21,9 @@ serde = { workspace = true, features = ["derive"] } async-trait = "0.1" tokio = "1.46" +[dev-dependencies] +tokio = { version = "1.46", features = ["macros", "rt-multi-thread"] } + [build-dependencies] uniffi = { version = "0.29", features = ["build"] } diff --git a/bindings/uniffi/nuget/README.md b/bindings/uniffi/nuget/README.md index 10c9a593..49c661a0 100644 --- a/bindings/uniffi/nuget/README.md +++ b/bindings/uniffi/nuget/README.md @@ -22,6 +22,18 @@ Console.WriteLine(response.result); ``` +## Loader Configurations + +The `loader` argument accepts either a callback (`ZenLoader.Callback`) or a loader configuration +of a known type. With a configuration, decisions are pre-loaded and pre-compiled at engine +creation for faster evaluations: + +```csharp +var fsEngine = new ZenEngine(loader: new ZenLoader.Filesystem("decisions")); +var zipEngine = new ZenEngine(loader: new ZenLoader.Zip(File.ReadAllBytes("decisions.zip"))); +var cbEngine = new ZenEngine(loader: new ZenLoader.Callback(new FileLoader())); +``` + ## Features - **Decision Tables** - Rule tables with first/collect hit policies diff --git a/bindings/uniffi/src/engine.rs b/bindings/uniffi/src/engine.rs index d00928f4..c6673c73 100644 --- a/bindings/uniffi/src/engine.rs +++ b/bindings/uniffi/src/engine.rs @@ -3,14 +3,13 @@ use crate::custom_node::{ }; use crate::decision::ZenDecision; use crate::error::ZenError; -use crate::loader::{ - NoopDecisionLoader, ZenDecisionLoaderCallback, ZenDecisionLoaderCallbackWrapper, -}; -use crate::types::{JsonBuffer, ZenEngineResponse}; +use crate::loader::{NoopDecisionLoader, ZenDecisionLoaderCallbackWrapper, ZenLoader}; +use crate::types::{JsonBuffer, ZenBatchRequest, ZenBatchResult, ZenEngineResponse}; use serde_json::Value; use std::sync::Arc; use tokio::runtime::Handle; use tokio::task; +use zen_engine::loader::DynamicLoader; use zen_engine::{DecisionEngine, EvaluationOptions}; #[derive(uniffi::Object)] @@ -44,21 +43,29 @@ impl From for EvaluationOptions { #[uniffi::export(async_runtime = "tokio")] impl ZenEngine { - #[uniffi::constructor] + #[uniffi::constructor(default(loader = None, custom_node = None))] pub fn new( - loader: Option>, + loader: Option, custom_node: Option>, - ) -> Self { - Self { - engine: Arc::new(DecisionEngine::new( - Arc::new(ZenDecisionLoaderCallbackWrapper( - loader.unwrap_or_else(|| Box::new(NoopDecisionLoader)), - )), - Arc::new(ZenCustomNodeCallbackWrapper( - custom_node.unwrap_or_else(|| Box::new(NoopCustomNodeCallback)), - )), + ) -> Result { + let loader: DynamicLoader = match loader { + Some(loader) => loader.into_dynamic_loader()?, + None => Arc::new(ZenDecisionLoaderCallbackWrapper(Arc::new( + NoopDecisionLoader, + ))), + }; + + let engine = DecisionEngine::new( + loader, + Arc::new(ZenCustomNodeCallbackWrapper( + custom_node.unwrap_or_else(|| Box::new(NoopCustomNodeCallback)), )), - } + ); + engine.compile(); + + Ok(Self { + engine: Arc::new(engine), + }) } pub async fn evaluate( @@ -94,6 +101,61 @@ impl ZenEngine { Ok(response) } + pub async fn evaluate_batch( + &self, + requests: Vec, + options: Option, + ) -> Vec { + let options: EvaluationOptions = options.unwrap_or_default().into(); + + let handles: Vec<_> = requests + .into_iter() + .map(|request| { + let engine = self.engine.clone(); + task::spawn_blocking(move || { + Handle::current().block_on(async move { + let context: Value = request.context.try_into()?; + let response = engine + .evaluate_with_opts(request.key, context.into(), options) + .await + .map_err(|err| { + ZenError::EvaluationError( + serde_json::to_string(&err.as_ref()) + .unwrap_or_else(|_| err.to_string()), + ) + })?; + + ZenEngineResponse::try_from(response) + }) + }) + }) + .collect(); + + let mut results = Vec::with_capacity(handles.len()); + for handle in handles { + let result = match handle.await { + Ok(Ok(data)) => ZenBatchResult { + success: true, + data: Some(data), + error: None, + }, + Ok(Err(error)) => ZenBatchResult { + success: false, + data: None, + error: Some(error.details()), + }, + Err(_) => ZenBatchResult { + success: false, + data: None, + error: Some("evaluation worker panicked".to_string()), + }, + }; + results.push(result); + } + + results + } + pub fn create_decision(&self, content: JsonBuffer) -> Result { let decision = self .engine @@ -131,3 +193,117 @@ impl ZenEngine { Ok(decision) } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::loader::ZenLoader; + use std::collections::HashMap; + + async fn assert_table_output(engine: &ZenEngine) { + let response = engine + .evaluate( + "table.json".to_string(), + JsonBuffer(br#"{"input":12}"#.to_vec()), + None, + ) + .await + .unwrap(); + + let result: Value = response.result.try_into().unwrap(); + assert_eq!(result["output"], serde_json::json!(10)); + } + + #[tokio::test(flavor = "multi_thread")] + async fn with_static_loader_config() { + let content = HashMap::from([( + "table.json".to_string(), + JsonBuffer(include_bytes!("../../../test-data/table.json").to_vec()), + )]); + + let engine = ZenEngine::new(Some(ZenLoader::Static { content }), None).unwrap(); + assert_table_output(&engine).await; + } + + #[tokio::test(flavor = "multi_thread")] + async fn with_filesystem_loader_config() { + let path = concat!(env!("CARGO_MANIFEST_DIR"), "/../../test-data").to_string(); + let engine = ZenEngine::new(Some(ZenLoader::Filesystem { path }), None).unwrap(); + assert_table_output(&engine).await; + } + + #[tokio::test(flavor = "multi_thread")] + async fn with_callback_loader() { + struct FsCallback; + + #[async_trait::async_trait] + impl crate::loader::ZenDecisionLoaderCallback for FsCallback { + async fn load(&self, key: String) -> Result, ZenError> { + let path = concat!(env!("CARGO_MANIFEST_DIR"), "/../../test-data"); + Ok(std::fs::read(format!("{path}/{key}")).ok().map(JsonBuffer)) + } + } + + let engine = ZenEngine::new( + Some(ZenLoader::Callback { + callback: Arc::new(FsCallback), + }), + None, + ) + .unwrap(); + assert_table_output(&engine).await; + } + + #[tokio::test(flavor = "multi_thread")] + async fn evaluate_batch_mixed_results() { + let path = concat!(env!("CARGO_MANIFEST_DIR"), "/../../test-data").to_string(); + let engine = ZenEngine::new(Some(ZenLoader::Filesystem { path }), None).unwrap(); + + let requests = vec![ + ZenBatchRequest { + key: "table.json".to_string(), + context: JsonBuffer(br#"{"input":12}"#.to_vec()), + }, + ZenBatchRequest { + key: "missing.json".to_string(), + context: JsonBuffer(b"{}".to_vec()), + }, + ZenBatchRequest { + key: "table.json".to_string(), + context: JsonBuffer(br#"{"input":5}"#.to_vec()), + }, + ]; + + let results = engine.evaluate_batch(requests, None).await; + assert_eq!(results.len(), 3); + + assert!(results[0].success); + let first: Value = + serde_json::from_slice(&results[0].data.as_ref().unwrap().result.0).unwrap(); + assert_eq!(first["output"], serde_json::json!(10)); + + assert!(!results[1].success); + assert!(results[1].error.is_some()); + + assert!(results[2].success); + let third: Value = + serde_json::from_slice(&results[2].data.as_ref().unwrap().result.0).unwrap(); + assert_eq!(third["output"], serde_json::json!(0)); + } + + #[tokio::test(flavor = "multi_thread")] + async fn with_static_loader_config_missing_key() { + let engine = ZenEngine::new( + Some(ZenLoader::Static { + content: HashMap::new(), + }), + None, + ) + .unwrap(); + + let result = engine + .evaluate("missing.json".to_string(), JsonBuffer(b"{}".to_vec()), None) + .await; + assert!(result.is_err()); + } +} diff --git a/bindings/uniffi/src/loader.rs b/bindings/uniffi/src/loader.rs index 7ab77f0a..bb92f809 100644 --- a/bindings/uniffi/src/loader.rs +++ b/bindings/uniffi/src/loader.rs @@ -1,14 +1,17 @@ use crate::error::ZenError; use crate::types::JsonBuffer; +use std::collections::HashMap; use std::fmt::{Debug, Formatter}; use std::future::Future; use std::pin::Pin; use std::sync::Arc; use uniffi::deps::anyhow::anyhow; -use zen_engine::loader::{DecisionLoader, LoaderError, LoaderResponse}; +use zen_engine::loader::{ + DecisionLoader, DynamicLoader, LoaderConfig, LoaderError, LoaderResponse, +}; use zen_engine::model::DecisionContent; -#[uniffi::export(callback_interface)] +#[uniffi::export(with_foreign)] #[async_trait::async_trait] pub trait ZenDecisionLoaderCallback: Send + Sync { async fn load(&self, key: String) -> Result, ZenError>; @@ -23,7 +26,52 @@ impl ZenDecisionLoaderCallback for NoopDecisionLoader { } } -pub struct ZenDecisionLoaderCallbackWrapper(pub Box); +#[derive(uniffi::Enum)] +pub enum ZenLoader { + Callback { + callback: Arc, + }, + Static { + content: HashMap, + }, + Filesystem { + path: String, + }, + Zip { + bytes: Vec, + }, +} + +impl ZenLoader { + pub fn into_dynamic_loader(self) -> Result { + let config = match self { + ZenLoader::Callback { callback } => { + return Ok(Arc::new(ZenDecisionLoaderCallbackWrapper(callback))); + } + ZenLoader::Static { content } => { + let content = content + .into_iter() + .map(|(key, buffer)| { + let decision_content: DecisionContent = + serde_json::from_slice(buffer.0.as_slice()) + .map_err(|_| ZenError::JsonDeserializationFailed)?; + Ok((key, decision_content)) + }) + .collect::, ZenError>>()?; + + LoaderConfig::Static { content } + } + ZenLoader::Filesystem { path } => LoaderConfig::Filesystem { path }, + ZenLoader::Zip { bytes } => LoaderConfig::Zip { bytes }, + }; + + config + .into_loader() + .map_err(|e| ZenError::ValidationError(e.to_string())) + } +} + +pub struct ZenDecisionLoaderCallbackWrapper(pub Arc); impl Debug for ZenDecisionLoaderCallbackWrapper { fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { diff --git a/bindings/uniffi/src/types.rs b/bindings/uniffi/src/types.rs index 79bf884f..a1843859 100644 --- a/bindings/uniffi/src/types.rs +++ b/bindings/uniffi/src/types.rs @@ -98,6 +98,19 @@ impl TryFrom for ZenEngineResponse { } } +#[derive(uniffi::Record)] +pub struct ZenBatchRequest { + pub key: String, + pub context: JsonBuffer, +} + +#[derive(uniffi::Record)] +pub struct ZenBatchResult { + pub success: bool, + pub data: Option, + pub error: Option, +} + #[derive(uniffi::Record)] pub struct ZenEngineHandlerResponse { pub output: JsonBuffer, diff --git a/core/engine/src/engine.rs b/core/engine/src/engine.rs index f92af51f..9d85018a 100644 --- a/core/engine/src/engine.rs +++ b/core/engine/src/engine.rs @@ -34,7 +34,7 @@ impl Debug for DecisionEngine { } } -#[derive(Debug)] +#[derive(Debug, Clone, Copy)] pub struct EvaluationOptions { pub trace: bool, pub max_depth: u8,