diff --git a/crates/executor-core/src/engine.rs b/crates/executor-core/src/engine.rs index 5e26879..d5660b0 100644 --- a/crates/executor-core/src/engine.rs +++ b/crates/executor-core/src/engine.rs @@ -295,6 +295,10 @@ impl Engine { // fan-out is counted across the whole synchronous chain. Held to the end // of the call; its Drop re-zeroes the counter on the outermost exit. let _emit_scope = crate::sdk::emit_budget::EmissionBudgetScope::enter(); + // §9.4 M6: memoize interceptor chain resolution for this execution tree + // (same re-entrancy model as the emission budget); cleared at the + // outermost boundary so a pooled thread never reuses a foreign app's cache. + let _interceptor_cache = crate::sdk::interceptor::InterceptorCacheScope::enter(); let effective_limits = self.limits.with_overrides(&req.sandbox_overrides); let logs: Arc>> = Arc::new(Mutex::new(Vec::new())); let mut engine = build_engine(effective_limits, Some(logs.clone())); diff --git a/crates/executor-core/src/sdk/interceptor.rs b/crates/executor-core/src/sdk/interceptor.rs index 97b26a3..8ef66be 100644 --- a/crates/executor-core/src/sdk/interceptor.rs +++ b/crates/executor-core/src/sdk/interceptor.rs @@ -23,10 +23,13 @@ //! execution, makes any nested op the interceptor performs bypass interception. use std::cell::{Cell, RefCell}; +use std::collections::HashMap; use std::sync::{Arc, LazyLock}; use std::time::Duration; -use picloud_shared::{InterceptorChain, InterceptorEntry, InterceptorService, ScriptId, SdkCallCx}; +use picloud_shared::{ + AppId, InterceptorChain, InterceptorEntry, InterceptorService, ScriptId, SdkCallCx, +}; use rhai::EvalAltResult; use serde_json::{json, Value as Json}; use tokio::runtime::Handle as TokioHandle; @@ -72,6 +75,51 @@ thread_local! { /// `IN_INTERCEPTOR` counter, which suppresses interception of an /// interceptor's OWN nested writes but does not track identity. static VISITED: RefCell> = const { RefCell::new(Vec::new()) }; + + /// §9.4 M6: per-execution resolve cache. Keyed by `(app_id, service, op)` — + /// the chain a `(service, op)` resolves to is constant for one app across an + /// execution tree, so the FIRST hooked-eligible op of that kind pays the one + /// chain query and every later op reuses it. The dominant win is the + /// un-hooked case: an empty chain is cached once, so N `kv::set`s in a script + /// with no interceptors issue ONE resolve, not N. Cleared at the outermost + /// execution boundary so a pooled thread never serves a stale (or foreign) + /// app's cache. + static RESOLVE_CACHE: RefCell> = + RefCell::new(HashMap::new()); + /// Scope nesting depth for the resolve cache (mirrors the emission budget): + /// `0` when no execution is active on this thread. + static CACHE_ACTIVE: Cell = const { Cell::new(0) }; +} + +/// RAII scope wrapping one `execute_ast` for the §9.4 M6 resolve cache. +/// Re-entrancy aware: the OUTERMOST scope on a thread clears the cache on entry +/// and on final exit; a nested synchronous re-entry (invoke / interceptor) +/// shares the same cache, so a whole execution tree resolves each `(service, +/// op)` at most once. +pub(crate) struct InterceptorCacheScope; + +impl InterceptorCacheScope { + pub(crate) fn enter() -> Self { + CACHE_ACTIVE.with(|a| { + if a.get() == 0 { + RESOLVE_CACHE.with(|c| c.borrow_mut().clear()); + } + a.set(a.get() + 1); + }); + Self + } +} + +impl Drop for InterceptorCacheScope { + fn drop(&mut self) { + CACHE_ACTIVE.with(|a| { + let n = a.get().saturating_sub(1); + a.set(n); + if n == 0 { + RESOLVE_CACHE.with(|c| c.borrow_mut().clear()); + } + }); + } } fn currently_in_interceptor() -> bool { @@ -205,21 +253,29 @@ pub(super) fn run_after( } /// Resolve the before+after chain for `(service, op)` — every marker visible on -/// the calling app's chain, ordered per phase. An `Err` (no tokio runtime, or a -/// backend failure) makes the caller fail closed rather than silently allow. +/// the calling app's chain, ordered per phase. Memoized per execution tree +/// (§9.4 M6): the FIRST call for a `(app_id, service, op)` pays the chain query; +/// later calls reuse the cached result. An `Err` (no tokio runtime, or a backend +/// failure) makes the caller fail closed rather than silently allow. fn resolve_chain( ictx: &InterceptorCtx, cx: &Arc, service: &'static str, op: &'static str, ) -> Result> { + let cache_key = (cx.app_id, service.to_string(), op.to_string()); + if let Some(hit) = RESOLVE_CACHE.with(|c| c.borrow().get(&cache_key).cloned()) { + return Ok(hit); + } let handle = TokioHandle::try_current() .map_err(|e| runtime_err(&format!("{service} interceptor: no tokio runtime: {e}")))?; let interceptors = ictx.interceptors.clone(); - let cx = cx.clone(); - handle - .block_on(async move { interceptors.resolve(&cx, service, op).await }) - .map_err(|e| runtime_err(&format!("{service}::{op} interceptor resolve: {e}"))) + let resolve_cx = cx.clone(); + let chain = handle + .block_on(async move { interceptors.resolve(&resolve_cx, service, op).await }) + .map_err(|e| runtime_err(&format!("{service}::{op} interceptor resolve: {e}")))?; + RESOLVE_CACHE.with(|c| c.borrow_mut().insert(cache_key, chain.clone())); + Ok(chain) } /// Build the operation-context payload handed to a hook. `result` is `Some` only diff --git a/crates/executor-core/tests/sdk_kv.rs b/crates/executor-core/tests/sdk_kv.rs index a0f8cd9..5a4659b 100644 --- a/crates/executor-core/tests/sdk_kv.rs +++ b/crates/executor-core/tests/sdk_kv.rs @@ -329,3 +329,81 @@ async fn kv_bridge_preserves_cross_app_isolation() { by app_id, not execution_id/script_id" ); } + +// --- §9.4 M6: per-execution interceptor resolve cache --------------------- + +/// An `InterceptorService` that resolves to an EMPTY chain (nothing hooked) but +/// counts every `resolve` call, so a test can prove the per-execution cache +/// collapses N same-`(service, op)` resolves into one. +#[derive(Default)] +struct CountingInterceptors { + calls: std::sync::atomic::AtomicUsize, +} + +#[async_trait] +impl picloud_shared::InterceptorService for CountingInterceptors { + async fn resolve( + &self, + _cx: &SdkCallCx, + _service: &str, + _op: &str, + ) -> Result { + self.calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + Ok(picloud_shared::InterceptorChain::default()) + } +} + +fn make_engine_with_interceptors(ic: Arc) -> Arc { + let services = Services::new( + Arc::new(InMemoryKv::default()), + Arc::new(NoopDocsService), + Arc::new(NoopDeadLetterService), + Arc::new(NoopEventEmitter), + Arc::new(NoopModuleSource), + Arc::new(NoopHttpService), + Arc::new(picloud_shared::NoopFilesService), + Arc::new(picloud_shared::NoopPubsubService), + Arc::new(picloud_shared::NoopSecretsService), + Arc::new(picloud_shared::NoopEmailService), + Arc::new(picloud_shared::NoopUsersService), + Arc::new(picloud_shared::NoopQueueService), + Arc::new(picloud_shared::NoopInvokeService), + Arc::new(picloud_shared::NoopVarsService), + ) + .with_interceptors(ic); + Arc::new(Engine::new(Limits::default(), services)) +} + +/// M6: within ONE execution, N `kv::set`s of the same `(service, op)` resolve +/// the interceptor chain at most once (the empty chain is cached), and a SECOND +/// execution starts fresh (the cache is cleared at the outermost boundary). +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn resolve_is_cached_per_execution() { + let ic = Arc::new(CountingInterceptors::default()); + let engine = make_engine_with_interceptors(ic.clone()); + let app = AppId::new(); + + // 25 sets in one execution — all the same (kv, set). + let src = r#" + let c = kv::collection("c"); + let n = 0; + while n < 25 { c.set("k" + n, n); n += 1; } + "done" + "#; + let body = run_script(engine.clone(), src, baseline_request(app)).await; + assert_eq!(body, Value::from("done")); + assert_eq!( + ic.calls.load(std::sync::atomic::Ordering::SeqCst), + 1, + "25 kv::sets in one execution must resolve the (kv, set) chain exactly once" + ); + + // A second execution resolves again (the cache cleared at the outermost exit). + let body = run_script(engine, src, baseline_request(app)).await; + assert_eq!(body, Value::from("done")); + assert_eq!( + ic.calls.load(std::sync::atomic::Ordering::SeqCst), + 2, + "a fresh execution must not reuse the previous execution's resolve cache" + ); +}