From 302c1df577bdb54b7acdb11da38373749b9abcb0 Mon Sep 17 00:00:00 2001 From: MechaCat02 Date: Sat, 6 Jun 2026 19:56:05 +0200 Subject: [PATCH] =?UTF-8?q?feat(v1.1.9):=20invoke::=20Rhai=20SDK=20module?= =?UTF-8?q?=20=E2=80=94=20sync=20re-entry=20+=20invoke=5Fasync?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit executor-core/sdk/invoke.rs adds `invoke()` and `invoke_async()` as top-level Rhai helpers (no `::` namespace, mirroring the brief): invoke(target, args) -> Dynamic (sync, returns callee's value) invoke_async(target, args) -> String (fire-and-forget; returns execution_id) Sync re-entry pattern: - bridge captures Arc via Engine::self_arc() - calls engine.execute(&source, req) directly — same engine instance, same Services, same Limits; only SdkCallCx changes per call - runs inside the caller's spawn_blocking thread (no nested spawn_blocking, no gate re-admission — the outer execution already holds a permit) Back-reference plumbing: - Engine gains self_weak: OnceLock> + set_self_weak() + self_arc() accessor - picloud binary calls engine.set_self_weak(Arc::downgrade(&engine)) right after construction - register_all extended with limits + Option> params - Engine::execute_ast threads both through Limits.trigger_depth_max added — mirrors TriggerConfig::max_trigger_depth. Default 8; the picloud binary syncs from TriggerConfig::from_env() so PICLOUD_MAX_TRIGGER_DEPTH governs both the dispatcher's fan-out cap and the invoke depth bound. Target parsing (string-first): - "/api/foo" → InvokeTarget::Path - 36-char UUID-like → InvokeTarget::Id (uuid::Uuid parse-validated) - everything else → InvokeTarget::Name Cross-app + depth + FnPtr guards: - resolve() returns InvokeError::CrossApp if resolved.app_id != cx.app_id - bridge throws "invoke: depth limit exceeded (max N)" when cx.trigger_depth + 1 > limits.trigger_depth_max (checked BEFORE resolve to avoid a wasted DB round-trip) - args_to_json rejects FnPtr at any depth — closures don't survive invoke boundaries invoke_async path: - calls services.invoke.enqueue_async() which writes an OutboxSourceKind::Invoke row (commit 10 wires the dispatcher arm) - returns the new ExecutionId as a string for caller tracking Integration tests (sdk_invoke.rs, 6 binaries): - sync invoke returns callee's value (script returns body.x + 1) - cross-app invoke rejected (target in other app) - depth limit exceeded (recursive script that calls itself; throws before stack overflow) - callee receives incremented depth in cx (smoke — strict assertion needs ctx.trigger_depth surface, deferred to v1.2) - args_to_json rejects FnPtr in payload - invoke_async returns valid UUID string + queues args payload Co-Authored-By: Claude Opus 4.7 (1M context) --- crates/executor-core/src/engine.rs | 41 ++- crates/executor-core/src/sandbox.rs | 14 + crates/executor-core/src/sdk/invoke.rs | 330 +++++++++++++++++++++++ crates/executor-core/src/sdk/mod.rs | 21 +- crates/executor-core/tests/sdk_invoke.rs | 248 +++++++++++++++++ crates/picloud/src/lib.rs | 10 +- 6 files changed, 657 insertions(+), 7 deletions(-) create mode 100644 crates/executor-core/src/sdk/invoke.rs create mode 100644 crates/executor-core/tests/sdk_invoke.rs diff --git a/crates/executor-core/src/engine.rs b/crates/executor-core/src/engine.rs index aa23581..ce7a835 100644 --- a/crates/executor-core/src/engine.rs +++ b/crates/executor-core/src/engine.rs @@ -1,5 +1,5 @@ use std::collections::BTreeMap; -use std::sync::{Arc, Mutex}; +use std::sync::{Arc, Mutex, OnceLock, Weak}; use std::time::Instant; use chrono::Utc; @@ -44,6 +44,14 @@ pub struct Engine { /// `(app_id, name)`; invalidated lazily by `updated_at` mismatch /// at resolver time. module_cache: Arc, + /// v1.1.9: back-reference set by the picloud binary after + /// `Arc::new(Engine::new(...))`. The `invoke` SDK bridge reads it + /// to re-enter the engine synchronously for `invoke()`. `Weak` so + /// holding the Engine in an Arc doesn't create a strong cycle. + /// `None` until `set_self_weak` runs; the bridge surfaces a clear + /// error if invoke is called without the back-reference being set + /// (which only happens in tests that don't wire it). + self_weak: OnceLock>, } impl Engine { @@ -67,9 +75,31 @@ impl Engine { limits, services, module_cache: new_module_cache(module_cache_capacity), + self_weak: OnceLock::new(), } } + /// v1.1.9: install the back-reference used by the `invoke` SDK + /// bridge for synchronous re-entry. Idempotent (subsequent calls + /// are no-ops). The picloud binary calls this right after + /// `Arc::new(Engine::new(...))`: + /// + /// ```ignore + /// let engine = Arc::new(Engine::new(limits, services)); + /// engine.set_self_weak(Arc::downgrade(&engine)); + /// ``` + pub fn set_self_weak(&self, weak: Weak) { + let _ = self_weak_set(&self.self_weak, weak); + } + + /// Internal accessor used by the `invoke` SDK bridge — returns + /// `Some(strong)` if the back-reference was installed and the + /// engine is still alive. + #[must_use] + pub fn self_arc(&self) -> Option> { + self.self_weak.get().and_then(Weak::upgrade) + } + #[must_use] pub fn limits(&self) -> &Limits { &self.limits @@ -164,7 +194,8 @@ impl Engine { effective_limits.module_import_depth_max, ); engine.set_module_resolver(resolver); - sdk::register_all(&mut engine, &self.services, cx); + let self_engine = self.self_arc(); + sdk::register_all(&mut engine, &self.services, cx, effective_limits, self_engine); let mut scope = Scope::new(); scope.push_constant("ctx", build_ctx_map(&req)); @@ -211,6 +242,12 @@ impl ScriptValidator for Engine { } } +/// Tiny helper to make the `set_self_weak` body idempotent without +/// pulling the impl up into the public API. +fn self_weak_set(slot: &OnceLock>, weak: Weak) -> Result<(), Weak> { + slot.set(weak) +} + // ---------------------------------------------------------------------------- // Engine construction // ---------------------------------------------------------------------------- diff --git a/crates/executor-core/src/sandbox.rs b/crates/executor-core/src/sandbox.rs index 5ce8061..00cf227 100644 --- a/crates/executor-core/src/sandbox.rs +++ b/crates/executor-core/src/sandbox.rs @@ -30,6 +30,16 @@ pub struct Limits { /// Not script-overridable (this is a platform-level guard, not a /// per-script knob). pub module_import_depth_max: u32, + + /// v1.1.9: hard ceiling on `cx.trigger_depth` (shared between + /// trigger fan-out and `invoke()` re-entry). The dispatcher uses + /// `TriggerConfig::max_trigger_depth` for the SAME bound; the + /// invoke bridge uses this Limits-side mirror to short-circuit + /// before a DB round-trip. Defaults to 8 (matches + /// TriggerConfig::conservative); the picloud binary keeps the two + /// in sync by passing `TriggerConfig::from_env().max_trigger_depth` + /// through. + pub trigger_depth_max: u32, } impl Default for Limits { @@ -42,6 +52,7 @@ impl Default for Limits { max_call_levels: 64, max_expr_depth: 64, module_import_depth_max: 8, + trigger_depth_max: 8, } } } @@ -75,6 +86,9 @@ impl Limits { // module_import_depth_max is platform-level — overrides // never touch it. Carry through unchanged. module_import_depth_max: self.module_import_depth_max, + // trigger_depth_max is also platform-level (v1.1.9 invoke + // depth bound mirrors the dispatcher's trigger-depth cap). + trigger_depth_max: self.trigger_depth_max, } } } diff --git a/crates/executor-core/src/sdk/invoke.rs b/crates/executor-core/src/sdk/invoke.rs new file mode 100644 index 0000000..b28c8e5 --- /dev/null +++ b/crates/executor-core/src/sdk/invoke.rs @@ -0,0 +1,330 @@ +//! `invoke()` Rhai bridge — synchronous, same-app, re-entrant. +//! +//! ```rhai +//! let result = invoke("/api/payments/process", #{ order_id: 123 }); +//! let result = invoke(script_id("uuid-..."), args); +//! ``` +//! +//! Re-entrancy: the bridge captures `Arc` (via +//! `Engine::self_arc`) and calls `engine.execute(&source, req)` directly +//! inside the caller's `spawn_blocking` thread. Same engine instance, +//! same `Services`, same `Limits` — only the per-call `SdkCallCx` +//! changes (the callee gets `trigger_depth + 1` and a fresh +//! `execution_id`). +//! +//! Cross-app rejection happens at the `InvokeService::resolve` layer — +//! every method derives `app_id` from `cx.app_id` and returns +//! `InvokeError::CrossApp` if the resolved script belongs elsewhere. +//! +//! Depth bound: `cx.trigger_depth + 1 > limits.trigger_depth_max` → +//! `InvokeError::DepthExceeded`. Shared with the trigger fan-out depth +//! limit per the design notes (no separate `invoke_depth_max`). +//! +//! Closures captured across the `invoke` boundary are rejected at the +//! args-to-JSON conversion — `FnPtr` doesn't survive serialization and +//! re-entry into a fresh engine instance is the wrong scope for it +//! anyway. +//! +//! `invoke_async()` writes an outbox row (see `InvokeService::enqueue_async`) +//! and returns immediately with the new `ExecutionId` as a string. The +//! dispatcher fires it through the standard executor path; commit 10 +//! wires the dispatcher's `OutboxSourceKind::Invoke` arm. + +use std::collections::BTreeMap; +use std::sync::Arc; + +use base64::engine::general_purpose::STANDARD; +use base64::Engine as _; +use picloud_shared::{ExecutionId, InvokeService, InvokeTarget, ScriptId, SdkCallCx, Services}; +use rhai::{Array, Dynamic, Engine as RhaiEngine, EvalAltResult, Map, Module}; +use serde_json::Value as Json; +use tokio::runtime::Handle as TokioHandle; + +use crate::engine::Engine; +use crate::sandbox::Limits; +use crate::sdk::bridge::json_to_dynamic; +use crate::types::{ExecRequest, InvocationType}; + +pub(super) fn register( + engine: &mut RhaiEngine, + services: &Services, + cx: Arc, + limits: Limits, + self_engine: Option>, +) { + let svc = services.invoke.clone(); + let mut module = Module::new(); + + // Sync `invoke(target, args)` — returns the callee's return value. + { + let svc = svc.clone(); + let cx = cx.clone(); + let self_engine = self_engine.clone(); + module.set_native_fn( + "invoke", + move |target: Dynamic, args: Dynamic| -> Result> { + let target = parse_target(target)?; + let args_json = args_to_json(&args)?; + let self_engine = self_engine.clone().ok_or_else(|| -> Box { + EvalAltResult::ErrorRuntime( + "invoke: engine back-reference not installed (test setup missing set_self_weak?)".into(), + rhai::Position::NONE, + ) + .into() + })?; + invoke_blocking(&svc, &cx, target, args_json, limits, &self_engine) + }, + ); + } + + // `invoke_async(target, args)` — fire-and-forget; returns the new + // ExecutionId as a string. + { + let svc = svc.clone(); + let cx = cx.clone(); + module.set_native_fn( + "invoke_async", + move |target: Dynamic, args: Dynamic| -> Result> { + let target = parse_target(target)?; + let args_json = args_to_json(&args)?; + let svc = svc.clone(); + let cx = cx.clone(); + let handle = TokioHandle::try_current().map_err(|e| -> Box { + EvalAltResult::ErrorRuntime( + format!("invoke_async: no tokio runtime: {e}").into(), + rhai::Position::NONE, + ) + .into() + })?; + let exec_id = handle + .block_on(async move { svc.enqueue_async(&cx, target, args_json).await }) + .map_err(|e| -> Box { + EvalAltResult::ErrorRuntime( + format!("invoke_async: {e}").into(), + rhai::Position::NONE, + ) + .into() + })?; + Ok(exec_id.to_string()) + }, + ); + } + + // Register globally — `invoke(...)` and `invoke_async(...)` are + // top-level helpers, not under a `::` namespace, mirroring the + // brief's syntax. + engine.register_global_module(module.into()); +} + +/// The sync invoke path. Runs entirely inside the caller's +/// `spawn_blocking` thread; no gate (the outer execution is already +/// gate-admitted), no new spawn_blocking (we'd deadlock if we nested +/// inside the blocking pool). +fn invoke_blocking( + svc: &Arc, + cx: &Arc, + target: InvokeTarget, + args_json: Json, + limits: Limits, + self_engine: &Arc, +) -> Result> { + let handle = TokioHandle::try_current().map_err(|e| -> Box { + EvalAltResult::ErrorRuntime( + format!("invoke: no tokio runtime: {e}").into(), + rhai::Position::NONE, + ) + .into() + })?; + + // Depth check — done BEFORE resolve so a depth-exceeded loop + // doesn't waste a DB round-trip. + if cx.trigger_depth + 1 > limits.trigger_depth_max { + return Err(EvalAltResult::ErrorRuntime( + format!( + "invoke: depth limit exceeded (max {})", + limits.trigger_depth_max + ) + .into(), + rhai::Position::NONE, + ) + .into()); + } + + let svc_clone = svc.clone(); + let cx_clone = cx.clone(); + let target_label = target.describe(); + let resolved = handle + .block_on(async move { svc_clone.resolve(&cx_clone, target).await }) + .map_err(|e| -> Box { + EvalAltResult::ErrorRuntime( + format!("invoke({target_label}): {e}").into(), + rhai::Position::NONE, + ) + .into() + })?; + + let execution_id = ExecutionId::new(); + let req = ExecRequest { + execution_id, + request_id: cx.request_id, + script_id: resolved.script_id, + script_name: resolved.name.clone(), + invocation_type: InvocationType::Function, + path: "/invoke".into(), + headers: BTreeMap::new(), + body: args_json, + params: BTreeMap::new(), + query: BTreeMap::new(), + rest: String::new(), + sandbox_overrides: picloud_shared::ScriptSandbox::default(), + app_id: cx.app_id, + // Same-app invoke is a function call, not a re-auth boundary — + // inherit the caller's principal. + principal: cx.principal.clone(), + trigger_depth: cx.trigger_depth + 1, + root_execution_id: cx.root_execution_id, + is_dead_letter_handler: cx.is_dead_letter_handler, + event: None, + }; + + // Synchronous re-entry — same engine instance, same Services, + // fresh SdkCallCx built inside Engine::execute_ast. + let resp = self_engine + .execute(&resolved.source, req) + .map_err(|e| -> Box { + EvalAltResult::ErrorRuntime( + format!("invoke({target_label}): {e}").into(), + rhai::Position::NONE, + ) + .into() + })?; + + // The callee's return is the response `body` JSON. Convert back to + // Dynamic for the caller. Status code + headers are dropped; the + // function-call mental model is "return value", not HTTP response. + Ok(json_to_dynamic(resp.body)) +} + +/// Accept a string (route path OR script name) or a Rhai script-id +/// custom type. The string heuristic: starts with '/' → Path, else Name. +fn parse_target(target: Dynamic) -> Result> { + if let Ok(s) = target.clone().into_string() { + if s.starts_with('/') { + return Ok(InvokeTarget::Path(s)); + } + // Heuristic: 36-char UUID-like string → Id; else Name. Keeps + // `invoke("payments_worker")` and `invoke("550e8400-...")` both + // working without a separate `script_id()` constructor in v1.1.9. + if s.len() == 36 { + if let Ok(uuid) = uuid::Uuid::parse_str(&s) { + return Ok(InvokeTarget::Id(ScriptId::from(uuid))); + } + } + return Ok(InvokeTarget::Name(s)); + } + let _ = target; + Err(EvalAltResult::ErrorRuntime( + "invoke: target must be a string (path or name) or a script_id".into(), + rhai::Position::NONE, + ) + .into()) +} + +/// Convert Rhai Dynamic → JSON. Rejects FnPtr at any depth (closures +/// don't survive invoke boundaries). +fn args_to_json(value: &Dynamic) -> Result> { + if value.is::() { + return Err(EvalAltResult::ErrorRuntime( + "invoke: args must not contain FnPtr / closures".into(), + rhai::Position::NONE, + ) + .into()); + } + if value.is_blob() { + let blob = value.clone().into_blob().unwrap_or_default(); + return Ok(Json::String(STANDARD.encode(&blob))); + } + if value.is_unit() { + return Ok(Json::Null); + } + if let Ok(b) = value.as_bool() { + return Ok(Json::Bool(b)); + } + if let Ok(i) = value.as_int() { + return Ok(Json::Number(i.into())); + } + if let Ok(f) = value.as_float() { + return Ok(serde_json::Number::from_f64(f).map_or(Json::Null, Json::Number)); + } + if value.is_string() { + return Ok(Json::String(value.clone().into_string().unwrap_or_default())); + } + if let Some(arr) = value.clone().try_cast::() { + let mut out = Vec::with_capacity(arr.len()); + for v in &arr { + out.push(args_to_json(v)?); + } + return Ok(Json::Array(out)); + } + if let Some(map) = value.clone().try_cast::() { + let mut out = serde_json::Map::new(); + for (k, v) in map { + out.insert(k.to_string(), args_to_json(&v)?); + } + return Ok(Json::Object(out)); + } + Ok(Json::String(value.to_string())) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn parse_target_path() { + let d = Dynamic::from("/api/foo"); + let t = parse_target(d).unwrap(); + assert!(matches!(t, InvokeTarget::Path(_))); + } + + #[test] + fn parse_target_uuid_string_is_id() { + let d = Dynamic::from("550e8400-e29b-41d4-a716-446655440000"); + let t = parse_target(d).unwrap(); + assert!(matches!(t, InvokeTarget::Id(_))); + } + + #[test] + fn parse_target_plain_name() { + let d = Dynamic::from("payments_worker"); + let t = parse_target(d).unwrap(); + match t { + InvokeTarget::Name(n) => assert_eq!(n, "payments_worker"), + other => panic!("expected Name, got {other:?}"), + } + } + + #[test] + fn args_to_json_rejects_fnptr() { + let f = rhai::FnPtr::new("x").unwrap(); + let d = Dynamic::from(f); + assert!(args_to_json(&d).is_err()); + } + + #[test] + fn args_to_json_round_trips_map() { + let mut m = Map::new(); + m.insert("x".into(), Dynamic::from(42_i64)); + let d = Dynamic::from(m); + let j = args_to_json(&d).unwrap(); + assert_eq!(j, serde_json::json!({ "x": 42 })); + } +} + +/// `Limits` is `Copy`; passed by value into `register` so closures take +/// owned copies. Hint for readers wondering why it doesn't need `Arc`. +#[allow(dead_code)] +const _LIMITS_IS_COPY: fn() = || { + fn assert_copy() {} + assert_copy::(); +}; diff --git a/crates/executor-core/src/sdk/mod.rs b/crates/executor-core/src/sdk/mod.rs index a307435..d9f9443 100644 --- a/crates/executor-core/src/sdk/mod.rs +++ b/crates/executor-core/src/sdk/mod.rs @@ -18,6 +18,7 @@ pub mod docs; pub mod email; pub mod files; pub mod http; +pub mod invoke; pub mod kv; pub mod pubsub; pub mod queue; @@ -33,13 +34,24 @@ use std::sync::Arc; use picloud_shared::Services; use rhai::Engine as RhaiEngine; +use crate::engine::Engine; +use crate::sandbox::Limits; + /// Single hook every v1.1.x stateful service registers into. Called /// once per invocation, just after `build_engine` constructs the /// sandboxed Rhai engine and just before script compilation. /// -/// v1.1.1 wires the first stateful service (KV). Subsequent PRs add a -/// single `::register(...)` line per service. -pub fn register_all(engine: &mut RhaiEngine, services: &Services, cx: Arc) { +/// v1.1.9 adds the `limits` + `self_engine` parameters needed by the +/// `invoke` bridge for synchronous re-entry. `self_engine` is `None` +/// in harnesses that didn't call `Engine::set_self_weak` after +/// construction; the invoke bridge surfaces a clear error in that case. +pub fn register_all( + engine: &mut RhaiEngine, + services: &Services, + cx: Arc, + limits: Limits, + self_engine: Option>, +) { kv::register(engine, services, cx.clone()); docs::register(engine, services, cx.clone()); dead_letters::register(engine, services, cx.clone()); @@ -49,5 +61,6 @@ pub fn register_all(engine: &mut RhaiEngine, services: &Services, cx: Arc>, // (id, app, name, source) + async_payloads: Mutex>, +} + +impl FakeInvokeService { + fn register(&self, app: AppId, name: &str, source: &str) -> ScriptId { + let id = ScriptId::new(); + self.scripts + .lock() + .unwrap() + .push((id, app, name.to_string(), source.to_string())); + id + } +} + +#[async_trait] +impl InvokeService for FakeInvokeService { + async fn resolve( + &self, + cx: &SdkCallCx, + target: InvokeTarget, + ) -> Result { + let entries = self.scripts.lock().unwrap().clone(); + let hit = entries.iter().find(|(id, _app, name, _)| match &target { + InvokeTarget::Id(t) => t == id, + InvokeTarget::Name(t) => t == name, + InvokeTarget::Path(t) => t == name, // tests stash route paths as names + }); + let (id, app, name, source): (ScriptId, AppId, String, String) = hit + .ok_or_else(|| InvokeError::NotFound(target.describe()))? + .clone(); + if app != cx.app_id { + return Err(InvokeError::CrossApp); + } + Ok(ResolvedScript { + script_id: id, + app_id: app, + source, + updated_at: Utc::now(), + name, + }) + } + + async fn enqueue_async( + &self, + _cx: &SdkCallCx, + _target: InvokeTarget, + args: Value, + ) -> Result { + self.async_payloads.lock().unwrap().push(args); + Ok(ExecutionId::new()) + } +} + +fn build_engine(svc: Arc) -> Arc { + let services = Services::new( + Arc::new(NoopKvService), + Arc::new(NoopDocsService), + Arc::new(NoopDeadLetterService), + Arc::new(NoopEventEmitter), + Arc::new(NoopModuleSource), + Arc::new(NoopHttpService), + Arc::new(NoopFilesService), + Arc::new(NoopPubsubService), + Arc::new(NoopSecretsService), + Arc::new(NoopEmailService), + Arc::new(NoopUsersService), + Arc::new(NoopQueueService), + svc, + ); + let engine = Arc::new(Engine::new(Limits::default(), services)); + engine.set_self_weak(Arc::downgrade(&engine)); + engine +} + +fn baseline_request(app_id: AppId) -> ExecRequest { + let execution_id = ExecutionId::new(); + ExecRequest { + execution_id, + request_id: RequestId::new(), + script_id: ScriptId::new(), + script_name: "caller".into(), + invocation_type: InvocationType::Http, + path: "/caller".into(), + headers: BTreeMap::new(), + body: Value::Null, + params: BTreeMap::new(), + query: BTreeMap::new(), + rest: String::new(), + sandbox_overrides: ScriptSandbox::default(), + app_id, + principal: None, + trigger_depth: 0, + root_execution_id: execution_id, + is_dead_letter_handler: false, + event: None, + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn invoke_returns_callee_value() { + let app = AppId::new(); + let svc = Arc::new(FakeInvokeService::default()); + svc.register(app, "worker", "ctx.request.body.x + 1"); + let engine = build_engine(svc); + let src = r#" + let n = invoke("worker", #{ x: 41 }); + #{ statusCode: 200, body: n } + "# + .to_string(); + let req = baseline_request(app); + let resp = tokio::task::spawn_blocking(move || engine.execute(&src, req)) + .await + .unwrap() + .unwrap(); + assert_eq!(resp.body, json!(42)); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn invoke_cross_app_rejects() { + let caller_app = AppId::new(); + let other_app = AppId::new(); + let svc = Arc::new(FakeInvokeService::default()); + svc.register(other_app, "worker", "0"); // belongs to other_app + let engine = build_engine(svc); + let src = r#"invoke("worker", #{});"#.to_string(); + let req = baseline_request(caller_app); + let res = tokio::task::spawn_blocking(move || engine.execute(&src, req)) + .await + .unwrap(); + let err = res.unwrap_err().to_string(); + assert!( + err.contains("different app") || err.contains("CrossApp"), + "expected cross-app rejection, got: {err}" + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn invoke_depth_limit_exceeded() { + let app = AppId::new(); + let svc = Arc::new(FakeInvokeService::default()); + // recursive: calls itself again — would loop without the bound. + svc.register(app, "loop", r#"invoke("loop", #{})"#); + let engine = build_engine(svc); + let src = r#"invoke("loop", #{});"#.to_string(); + let req = baseline_request(app); + let res = tokio::task::spawn_blocking(move || engine.execute(&src, req)) + .await + .unwrap(); + let err = res.unwrap_err().to_string(); + assert!( + err.contains("depth limit"), + "expected depth limit error, got: {err}" + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn invoke_callee_sees_incremented_depth() { + let app = AppId::new(); + let svc = Arc::new(FakeInvokeService::default()); + // The callee returns its own trigger_depth so the caller can assert + // it bumped from 0 → 1. + svc.register(app, "depth_probe", "ctx.trigger_depth"); + let engine = build_engine(svc); + let src = r#" + let d = invoke("depth_probe", #{}); + #{ statusCode: 200, body: d } + "# + .to_string(); + let req = baseline_request(app); + let resp = tokio::task::spawn_blocking(move || engine.execute(&src, req)) + .await + .unwrap() + .unwrap(); + // ctx exposes trigger_depth? — not yet (it's not in build_ctx_map). + // Skip the strict assertion — the test still ensures the invoke + // chain didn't throw. (See HANDBACK §11 for cx.trigger_depth surface + // exposure as a v1.2 follow-up.) + let _ = resp; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn invoke_rejects_fnptr_in_args() { + let app = AppId::new(); + let svc = Arc::new(FakeInvokeService::default()); + svc.register(app, "worker", "0"); + let engine = build_engine(svc); + let src = r#" + let f = |x| x + 1; + invoke("worker", #{ cb: f }); + "# + .to_string(); + let req = baseline_request(app); + let res = tokio::task::spawn_blocking(move || engine.execute(&src, req)) + .await + .unwrap(); + let err = res.unwrap_err().to_string(); + assert!( + err.contains("FnPtr") || err.contains("closure"), + "expected FnPtr rejection, got: {err}" + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn invoke_async_returns_execution_id_string() { + let app = AppId::new(); + let svc = Arc::new(FakeInvokeService::default()); + svc.register(app, "worker", "0"); + let engine = build_engine(svc.clone()); + let src = r#" + let id = invoke_async("worker", #{ x: 1 }); + #{ statusCode: 200, body: id } + "# + .to_string(); + let req = baseline_request(app); + let resp = tokio::task::spawn_blocking(move || engine.execute(&src, req)) + .await + .unwrap() + .unwrap(); + // Body is a string — a UUID — surfaced as Json::String. + let id = resp.body.as_str().expect("body should be a string"); + assert!(uuid::Uuid::parse_str(id).is_ok(), "expected UUID, got: {id}"); + let payloads = svc.async_payloads.lock().unwrap().clone(); + assert_eq!(payloads.len(), 1); + assert_eq!(payloads[0], json!({ "x": 1 })); +} diff --git a/crates/picloud/src/lib.rs b/crates/picloud/src/lib.rs index 10b61ab..162b74f 100644 --- a/crates/picloud/src/lib.rs +++ b/crates/picloud/src/lib.rs @@ -310,7 +310,15 @@ pub async fn build_app( queue, invoke, ); - let engine = Arc::new(Engine::new(Limits::default(), services)); + // v1.1.9: keep the invoke depth bound aligned with the dispatcher's + // trigger-depth bound (same counter under the hood). + let mut engine_limits = Limits::default(); + engine_limits.trigger_depth_max = trigger_config.max_trigger_depth; + let engine = Arc::new(Engine::new(engine_limits, services)); + // v1.1.9: install the back-reference the `invoke` SDK bridge needs + // for synchronous re-entry. Weak so the Arc-cycle stays loose; + // OnceLock-backed so it's idempotent. + engine.set_self_weak(Arc::downgrade(&engine)); // Same shape for app domains (Host → app_id cache). let app_domain_table = Arc::new(AppDomainTable::new());