feat(v1.1.9): invoke:: Rhai SDK module — sync re-entry + invoke_async

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<Engine> 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<Weak<Engine>> + 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<Arc<Engine>> 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) <noreply@anthropic.com>
This commit is contained in:
MechaCat02
2026-06-06 19:56:05 +02:00
parent 563c44ad95
commit 302c1df577
6 changed files with 657 additions and 7 deletions

View File

@@ -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<Engine>` (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<SdkCallCx>,
limits: Limits,
self_engine: Option<Arc<Engine>>,
) {
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<Dynamic, Box<EvalAltResult>> {
let target = parse_target(target)?;
let args_json = args_to_json(&args)?;
let self_engine = self_engine.clone().ok_or_else(|| -> Box<EvalAltResult> {
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<String, Box<EvalAltResult>> {
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> {
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> {
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<dyn InvokeService>,
cx: &Arc<SdkCallCx>,
target: InvokeTarget,
args_json: Json,
limits: Limits,
self_engine: &Arc<Engine>,
) -> Result<Dynamic, Box<EvalAltResult>> {
let handle = TokioHandle::try_current().map_err(|e| -> Box<EvalAltResult> {
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> {
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> {
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<InvokeTarget, Box<EvalAltResult>> {
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<Json, Box<EvalAltResult>> {
if value.is::<rhai::FnPtr>() {
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::<Array>() {
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::<Map>() {
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<T: Copy>() {}
assert_copy::<Limits>();
};

View File

@@ -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 `<service>::register(...)` line per service.
pub fn register_all(engine: &mut RhaiEngine, services: &Services, cx: Arc<SdkCallCx>) {
/// 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<SdkCallCx>,
limits: Limits,
self_engine: Option<Arc<Engine>>,
) {
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<SdkCal
queue::register(engine, services, cx.clone());
secrets::register(engine, services, cx.clone());
email::register(engine, services, cx.clone());
users::register(engine, services, cx);
users::register(engine, services, cx.clone());
invoke::register(engine, services, cx, limits, self_engine);
}