fabi/PiCloud
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(interceptors): ordered before-chains + cycle guard + ctx plumbing (§9.4 M1+M2)

Browse Source 
M1: introduces a clone-cheap InterceptorCtx { interceptors, self_engine,
limits } threaded into every SDK register fn (kv/docs/files/queue/pubsub/http)
so the non-KV services carry the hook seam (unused until M7-M11); KvHandle
collapses its three fields into one ictx.

M2: replaces the nearest-only resolver with ordered before/after chains. The
trait becomes resolve(cx, service, op) -> InterceptorChain { before, after }
(migration 0074 adds a phase column + phase-aware unique indexes). The before
-chain runs ancestor->app (depth DESC) so a group compliance guard can't be
bypassed by a descendant; single-marker behavior is byte-identical to before.
An identity cycle guard (thread-local visited-set keyed by script_id) denies a
detected cycle, alongside the existing binary re-entrancy break. Fail-closed
verdict preserved (allow only on #{ allowed: true }; a Dangling entry or a
missing engine back-ref denies); app_id still derives from cx.app_id only.

after-chains resolve but stay unused until M3. Pinned by two new interceptor
journeys (ancestor->app chain ordering; self-referential no-recurse).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
MechaCat02
2026-07-15 22:54:51 +02:00
parent eae2ee08f1
commit be0c618672
15 changed files with 623 additions and 200 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
crates/executor-core/src/sdk/docs.rs
View File

@@ -30,6 +30,7 @@ use uuid::Uuid;
use super::bridge::{
block_on, dynamic_to_json_capped, json_to_dynamic, MAX_JSON_MATERIALIZE_BYTES,
};
use super::interceptor::InterceptorCtx;
/// Per-call handle captured by the Rhai SDK. Cheap to clone (two Arcs
/// plus an owned string).
@@ -38,6 +39,10 @@ pub struct DocsHandle {
collection: String,
service: Arc<dyn DocsService>,
cx: Arc<SdkCallCx>,
// §9.4 M1 plumbing: carried so `create`/`update`/`delete` can run a before/
// after hook in a later milestone (M8). Unused until then.
#[allow(dead_code)]
ictx: InterceptorCtx,
}
/// §11.6 shared-collection handle, returned by `docs::shared_collection(name)`.
@@ -49,9 +54,17 @@ pub struct GroupDocsHandle {
collection: String,
service: Arc<dyn GroupDocsService>,
cx: Arc<SdkCallCx>,
// §9.4 M1 plumbing (see `DocsHandle`). Unused until M8.
#[allow(dead_code)]
ictx: InterceptorCtx,
}
pub(super) fn register(engine: &mut RhaiEngine, services: &Services, cx: Arc<SdkCallCx>) {
pub(super) fn register(
engine: &mut RhaiEngine,
services: &Services,
cx: Arc<SdkCallCx>,
ictx: InterceptorCtx,
) {
let docs_service = services.docs.clone();
let group_docs_service = services.group_docs.clone();
@@ -59,6 +72,7 @@ pub(super) fn register(engine: &mut RhaiEngine, services: &Services, cx: Arc<Sdk
{
let docs_service = docs_service.clone();
let cx = cx.clone();
let ictx = ictx.clone();
module.set_native_fn(
"collection",
move |name: &str| -> Result<DocsHandle, Box<EvalAltResult>> {
@@ -69,6 +83,7 @@ pub(super) fn register(engine: &mut RhaiEngine, services: &Services, cx: Arc<Sdk
collection: name.to_string(),
service: docs_service.clone(),
cx: cx.clone(),
ictx: ictx.clone(),
})
},
);
@@ -76,6 +91,7 @@ pub(super) fn register(engine: &mut RhaiEngine, services: &Services, cx: Arc<Sdk
{
let group_docs_service = group_docs_service.clone();
let cx = cx.clone();
let ictx = ictx.clone();
module.set_native_fn(
"shared_collection",
move |name: &str| -> Result<GroupDocsHandle, Box<EvalAltResult>> {
@@ -86,6 +102,7 @@ pub(super) fn register(engine: &mut RhaiEngine, services: &Services, cx: Arc<Sdk
collection: name.to_string(),
service: group_docs_service.clone(),
cx: cx.clone(),
ictx: ictx.clone(),
})
},
);

19
crates/executor-core/src/sdk/files.rs
View File

@@ -24,6 +24,7 @@
use std::sync::Arc;
use super::bridge::block_on;
use super::interceptor::InterceptorCtx;
use picloud_shared::{
FileMeta, FileUpdate, FilesService, GroupFilesService, NewFile, SdkCallCx, Services,
};
@@ -36,6 +37,10 @@ pub struct FilesHandle {
collection: String,
service: Arc<dyn FilesService>,
cx: Arc<SdkCallCx>,
// §9.4 M1 plumbing: carried so `create`/`update`/`delete` can run a before/
// after hook in a later milestone (M9). Unused until then.
#[allow(dead_code)]
ictx: InterceptorCtx,
}
/// §11.6 shared-collection handle, returned by `files::shared_collection(name)`.
@@ -47,9 +52,17 @@ pub struct GroupFilesHandle {
collection: String,
service: Arc<dyn GroupFilesService>,
cx: Arc<SdkCallCx>,
// §9.4 M1 plumbing (see `FilesHandle`). Unused until M9.
#[allow(dead_code)]
ictx: InterceptorCtx,
}
pub(super) fn register(engine: &mut RhaiEngine, services: &Services, cx: Arc<SdkCallCx>) {
pub(super) fn register(
engine: &mut RhaiEngine,
services: &Services,
cx: Arc<SdkCallCx>,
ictx: InterceptorCtx,
) {
let files_service = services.files.clone();
let group_files_service = services.group_files.clone();
@@ -57,6 +70,7 @@ pub(super) fn register(engine: &mut RhaiEngine, services: &Services, cx: Arc<Sdk
{
let files_service = files_service.clone();
let cx = cx.clone();
let ictx = ictx.clone();
module.set_native_fn(
"collection",
move |name: &str| -> Result<FilesHandle, Box<EvalAltResult>> {
@@ -67,6 +81,7 @@ pub(super) fn register(engine: &mut RhaiEngine, services: &Services, cx: Arc<Sdk
collection: name.to_string(),
service: files_service.clone(),
cx: cx.clone(),
ictx: ictx.clone(),
})
},
);
@@ -74,6 +89,7 @@ pub(super) fn register(engine: &mut RhaiEngine, services: &Services, cx: Arc<Sdk
{
let group_files_service = group_files_service.clone();
let cx = cx.clone();
let ictx = ictx.clone();
module.set_native_fn(
"shared_collection",
move |name: &str| -> Result<GroupFilesHandle, Box<EvalAltResult>> {
@@ -84,6 +100,7 @@ pub(super) fn register(engine: &mut RhaiEngine, services: &Services, cx: Arc<Sdk
collection: name.to_string(),
service: group_files_service.clone(),
cx: cx.clone(),
ictx: ictx.clone(),
})
},
);

12
crates/executor-core/src/sdk/http.rs
View File

@@ -38,6 +38,7 @@ use rhai::{Dynamic, Engine as RhaiEngine, EvalAltResult, Map, Module};
use super::bridge::{
block_on, dynamic_to_json_capped, json_to_dynamic, runtime_err, MAX_JSON_MATERIALIZE_BYTES,
};
use super::interceptor::InterceptorCtx;
/// Bridge-side defaults (the service clamps server-side too). The
/// `MAX_*` ceilings stay `i64` because they're compared against the
@@ -50,7 +51,16 @@ const MAX_REDIRECTS: i64 = 10;
const ALLOWED_OPT_KEYS: [&str; 4] = ["headers", "timeout_ms", "follow_redirects", "max_redirects"];
pub(super) fn register(engine: &mut RhaiEngine, services: &Services, cx: Arc<SdkCallCx>) {
// `http` registers its verbs as module native fns (no handle struct), so there
// is nowhere to stash the interceptor ctx yet. The param is threaded uniformly
// from `register_all` (§9.4 M1) and consumed by M11, which adds the before/after
// hook around the outbound-request closures. Prefixed `_` until then.
pub(super) fn register(
engine: &mut RhaiEngine,
services: &Services,
cx: Arc<SdkCallCx>,
_ictx: InterceptorCtx,
) {
let svc = services.http.clone();
let mut module = Module::new();

212
crates/executor-core/src/sdk/interceptor.rs
View File

@@ -22,10 +22,10 @@
//! thread-local guard, set for the duration of the interceptor's synchronous
//! execution, makes any nested op the interceptor performs bypass interception.
use std::cell::Cell;
use std::cell::{Cell, RefCell};
use std::sync::Arc;
use picloud_shared::{InterceptorResolution, InterceptorService, SdkCallCx};
use picloud_shared::{InterceptorEntry, InterceptorService, ScriptId, SdkCallCx};
use rhai::EvalAltResult;
use serde_json::{json, Value as Json};
use tokio::runtime::Handle as TokioHandle;
@@ -35,12 +35,31 @@ use crate::sandbox::Limits;
use crate::sdk::bridge::runtime_err;
use crate::sdk::invoke::run_resolved_blocking;
/// Clone-cheap bundle of the §9.4 interceptor dependencies threaded into every
/// SDK service `register` fn (M1). Carries the resolver, the `invoke()` re-entry
/// engine back-reference, and the sandbox limits — everything `run_before`
/// (and, from M3, `run_after`) needs. A few `Arc`s plus a small `Limits` copy.
#[derive(Clone)]
pub(crate) struct InterceptorCtx {
pub interceptors: Arc<dyn InterceptorService>,
pub self_engine: Option<Arc<Engine>>,
pub limits: Limits,
}
thread_local! {
/// Re-entrancy depth: `> 0` while an interceptor script (and anything it
/// synchronously invokes) is executing on this thread. The whole re-entry
/// runs synchronously in the caller's `spawn_blocking` thread (same model
/// as `invoke()`), so a thread-local is sufficient and correct.
static IN_INTERCEPTOR: Cell<u32> = const { Cell::new(0) };
/// Identity cycle guard (M2): the `script_id`s of the interceptors currently
/// on the synchronous run stack. A chain that would run an interceptor whose
/// script is already executing is a cycle — we DENY rather than recurse.
/// Precise (keyed by script id) and independent of the binary
/// `IN_INTERCEPTOR` counter, which suppresses interception of an
/// interceptor's OWN nested writes but does not track identity.
static VISITED: RefCell<Vec<ScriptId>> = const { RefCell::new(Vec::new()) };
}
fn currently_in_interceptor() -> bool {
@@ -62,17 +81,37 @@ impl Drop for ReentryGuard {
}
}
/// Run the before-op interceptor for `(service, op)` if one is registered.
/// Returns `Ok(())` to allow the operation (un-hooked, or the interceptor
/// explicitly allowed it) or an `Err` runtime error to deny it (the caller must
/// NOT then perform the write). `value` is the payload being written (`None`
/// for delete).
/// RAII for the identity cycle guard: push a `script_id` on enter, pop on drop
/// (including the `?`/panic-unwind paths), so the visited set exactly tracks the
/// interceptors on the current synchronous run stack.
struct CycleGuard;
impl CycleGuard {
fn enter(id: ScriptId) -> Self {
VISITED.with(|v| v.borrow_mut().push(id));
Self
}
fn contains(id: ScriptId) -> bool {
VISITED.with(|v| v.borrow().contains(&id))
}
}
impl Drop for CycleGuard {
fn drop(&mut self) {
VISITED.with(|v| {
v.borrow_mut().pop();
});
}
}
/// Run the before-op interceptor CHAIN for `(service, op)` if any are
/// registered. Returns `Ok(())` to allow the operation (un-hooked, or every
/// before-interceptor explicitly allowed it) or an `Err` runtime error to deny
/// it (the caller must NOT then perform the write). A single deny
/// short-circuits — the write is blocked. `value` is the payload being written
/// (`None` for delete).
#[allow(clippy::too_many_arguments)]
pub(super) fn run_before(
interceptors: &Arc<dyn InterceptorService>,
self_engine: Option<&Arc<Engine>>,
ictx: &InterceptorCtx,
cx: &Arc<SdkCallCx>,
limits: Limits,
service: &'static str,
op: &'static str,
collection: &str,
@@ -89,47 +128,21 @@ pub(super) fn run_before(
let handle = TokioHandle::try_current()
.map_err(|e| runtime_err(&format!("{service} interceptor: no tokio runtime: {e}")))?;
// (1) Resolve the nearest interceptor, sealed to its declaring owner.
let resolution = {
let interceptors = interceptors.clone();
// (1) Resolve the whole before+after chain, each entry sealed to its
// declaring owner. `before` is ordered ancestor→app (an outer/group guard
// runs first). `after` is unused this milestone.
let chain = {
let interceptors = ictx.interceptors.clone();
let cx = cx.clone();
handle
.block_on(async move { interceptors.resolve_before(&cx, service, op).await })
.block_on(async move { interceptors.resolve(&cx, service, op).await })
.map_err(|e| runtime_err(&format!("{service}::{op} interceptor resolve: {e}")))?
};
let resolved = match resolution {
// Un-hooked: the ONLY path that allows without running anything.
InterceptorResolution::None => return Ok(()),
// A guard is declared but its script is missing/disabled — fail closed.
InterceptorResolution::Dangling(name) => {
return Err(runtime_err(&format!(
"{service}::{op} denied: interceptor script `{name}` is missing or disabled"
)));
}
InterceptorResolution::Run(r) => r,
};
// A guard is registered but the engine back-reference isn't installed (a
// bare test engine without `set_self_weak`): we cannot run it, so DENY —
// a resolvable guard we can't execute must not fail open. Production always
// installs the back-ref.
let Some(self_engine) = self_engine else {
return Err(runtime_err(&format!(
"{service}::{op} denied: interceptor `{}` cannot run (engine back-reference not installed)",
resolved.name
)));
};
// Depth bound (shared with invoke / trigger fan-out).
if cx.trigger_depth + 1 > limits.trigger_depth_max {
return Err(runtime_err(&format!(
"{service}::{op} interceptor `{}`: depth limit exceeded (max {})",
resolved.name, limits.trigger_depth_max
)));
// Un-hooked: the ONLY path that allows without running anything.
if chain.before.is_empty() {
return Ok(());
}
// (2) Run the sealed script with the operation context as its request body,
// under the re-entrancy guard so its own writes bypass interception.
let payload = json!({
"service": service,
"action": op,
@@ -139,35 +152,88 @@ pub(super) fn run_before(
"caller_script_id": cx.script_id.to_string(),
"caller_execution_id": cx.execution_id.to_string(),
});
let ret = {
let _guard = ReentryGuard::enter();
run_resolved_blocking(
self_engine,
cx,
&resolved,
payload,
&format!("{service}::{op} interceptor `{}`", resolved.name),
)?
};
// Fail-closed verdict: allow ONLY on an explicit `#{ allowed: true }`. A
// bare bool, a typo'd key, a non-bool `allowed`, a bare unit, or any other
// shape DENIES — a control whose job is to deny must not allow by omission.
if let Json::Object(m) = &ret {
if m.get("allowed") == Some(&Json::Bool(true)) {
return Ok(());
// (2) Run each before-interceptor in order; the FIRST deny short-circuits.
for entry in &chain.before {
let resolved = match entry {
// A guard is declared but its script is missing/disabled — fail closed.
InterceptorEntry::Dangling(name) => {
return Err(runtime_err(&format!(
"{service}::{op} denied: interceptor script `{name}` is missing or disabled"
)));
}
InterceptorEntry::Run(r) => &r.script,
};
// Identity cycle guard: if this interceptor's script is already on the
// run stack, running it again would recurse forever — DENY.
if CycleGuard::contains(resolved.script_id) {
return Err(runtime_err(&format!(
"{service}::{op} denied: interceptor cycle detected: `{}`",
resolved.name
)));
}
// A guard is registered but the engine back-reference isn't installed (a
// bare test engine without `set_self_weak`): we cannot run it, so DENY —
// a resolvable guard we can't execute must not fail open. Production
// always installs the back-ref.
let Some(self_engine) = ictx.self_engine.as_ref() else {
return Err(runtime_err(&format!(
"{service}::{op} denied: interceptor `{}` cannot run (engine back-reference not installed)",
resolved.name
)));
};
// Depth bound (shared with invoke / trigger fan-out).
if cx.trigger_depth + 1 > ictx.limits.trigger_depth_max {
return Err(runtime_err(&format!(
"{service}::{op} interceptor `{}`: depth limit exceeded (max {})",
resolved.name, ictx.limits.trigger_depth_max
)));
}
// Run the sealed script with the operation context as its request body,
// under the re-entrancy guard (its own writes bypass interception) and
// the identity cycle guard (its own script id is on the stack).
let ret = {
let _reentry = ReentryGuard::enter();
let _cycle = CycleGuard::enter(resolved.script_id);
run_resolved_blocking(
self_engine,
cx,
resolved,
payload.clone(),
&format!("{service}::{op} interceptor `{}`", resolved.name),
)?
};
// Fail-closed verdict: allow ONLY on an explicit `#{ allowed: true }`. A
// bare bool, a typo'd key, a non-bool `allowed`, a bare unit, or any
// other shape DENIES — a control whose job is to deny must not allow by
// omission.
match &ret {
// Explicit allow → this entry passes; continue to the next.
Json::Object(m) if m.get("allowed") == Some(&Json::Bool(true)) => {}
Json::Object(m) => {
let reason = m
.get("reason")
.and_then(Json::as_str)
.unwrap_or("denied by interceptor");
return Err(runtime_err(&format!(
"{service}::{op} denied by interceptor `{}`: {reason}",
resolved.name
)));
}
_ => {
return Err(runtime_err(&format!(
"{service}::{op} denied by interceptor `{}`: no allow verdict returned",
resolved.name
)));
}
}
let reason = m
.get("reason")
.and_then(Json::as_str)
.unwrap_or("denied by interceptor");
return Err(runtime_err(&format!(
"{service}::{op} denied by interceptor `{}`: {reason}",
resolved.name
)));
}
Err(runtime_err(&format!(
"{service}::{op} denied by interceptor `{}`: no allow verdict returned",
resolved.name
)))
// Every before-interceptor allowed.
Ok(())
}

47
crates/executor-core/src/sdk/kv.rs
View File

@@ -30,26 +30,23 @@
use std::sync::Arc;
use picloud_shared::{GroupKvService, InterceptorService, KvService, SdkCallCx, Services};
use picloud_shared::{GroupKvService, KvService, SdkCallCx, Services};
use rhai::{Array, Dynamic, Engine as RhaiEngine, EvalAltResult, Map, Module};
use super::bridge::{
block_on, dynamic_to_json_capped, json_to_dynamic, JsonSizeError, MAX_JSON_MATERIALIZE_BYTES,
};
use crate::engine::Engine;
use crate::sandbox::Limits;
use super::interceptor::InterceptorCtx;
/// Per-call handle captured by the Rhai SDK. Cheap to clone (a few Arcs plus an
/// owned string). Carries the §9.4 interceptor deps so `set`/`delete` can run a
/// owned string). Carries the §9.4 interceptor ctx so `set`/`delete` can run a
/// before-op allow/deny hook (the resolver + the `invoke()` re-entry engine).
#[derive(Clone)]
pub struct KvHandle {
collection: String,
service: Arc<dyn KvService>,
cx: Arc<SdkCallCx>,
interceptors: Arc<dyn InterceptorService>,
self_engine: Option<Arc<Engine>>,
limits: Limits,
ictx: InterceptorCtx,
}
/// §11.6 shared-collection handle, returned by `kv::shared_collection(name)`. A distinct
@@ -57,28 +54,24 @@ pub struct KvHandle {
/// collections never grow triggers) and a script's choice of private-vs-shared
/// scope is explicit. Routes through the `GroupKvService`, which resolves the
/// owning group from `cx.app_id` (the script never names a group). Carries the
/// §9.4 interceptor deps too so a `(kv, set/delete)` guard covers shared-
/// §9.4 interceptor ctx too so a `(kv, set/delete)` guard covers shared-
/// collection writes — otherwise the shared handle would be a silent bypass.
#[derive(Clone)]
pub struct GroupKvHandle {
collection: String,
service: Arc<dyn GroupKvService>,
cx: Arc<SdkCallCx>,
interceptors: Arc<dyn InterceptorService>,
self_engine: Option<Arc<Engine>>,
limits: Limits,
ictx: InterceptorCtx,
}
pub(super) fn register(
engine: &mut RhaiEngine,
services: &Services,
cx: Arc<SdkCallCx>,
limits: Limits,
self_engine: Option<Arc<Engine>>,
ictx: InterceptorCtx,
) {
let kv_service = services.kv.clone();
let group_kv_service = services.group_kv.clone();
let interceptors = services.interceptors.clone();
// `kv::collection(name)` / `kv::shared_collection(name)` — both constructors live in
// the `kv` static module so the script-visible calls are `kv::collection`
@@ -87,8 +80,7 @@ pub(super) fn register(
{
let kv_service = kv_service.clone();
let cx = cx.clone();
let interceptors = interceptors.clone();
let self_engine = self_engine.clone();
let ictx = ictx.clone();
module.set_native_fn(
"collection",
move |name: &str| -> Result<KvHandle, Box<EvalAltResult>> {
@@ -99,9 +91,7 @@ pub(super) fn register(
collection: name.to_string(),
service: kv_service.clone(),
cx: cx.clone(),
interceptors: interceptors.clone(),
self_engine: self_engine.clone(),
limits,
ictx: ictx.clone(),
})
},
);
@@ -109,8 +99,7 @@ pub(super) fn register(
{
let group_kv_service = group_kv_service.clone();
let cx = cx.clone();
let interceptors = interceptors.clone();
let self_engine = self_engine.clone();
let ictx = ictx.clone();
module.set_native_fn(
"shared_collection",
move |name: &str| -> Result<GroupKvHandle, Box<EvalAltResult>> {
@@ -121,9 +110,7 @@ pub(super) fn register(
collection: name.to_string(),
service: group_kv_service.clone(),
cx: cx.clone(),
interceptors: interceptors.clone(),
self_engine: self_engine.clone(),
limits,
ictx: ictx.clone(),
})
},
);
@@ -188,10 +175,8 @@ fn register_set(engine: &mut RhaiEngine) {
// §9.4 before-op interceptor (allow/deny). A denial errors here and
// the write below never runs.
super::interceptor::run_before(
&handle.interceptors,
handle.self_engine.as_ref(),
&handle.ictx,
&handle.cx,
handle.limits,
"kv",
"set",
&handle.collection,
@@ -244,10 +229,8 @@ fn register_delete(engine: &mut RhaiEngine) {
"delete",
|handle: &mut KvHandle, key: &str| -> Result<bool, Box<EvalAltResult>> {
super::interceptor::run_before(
&handle.interceptors,
handle.self_engine.as_ref(),
&handle.ictx,
&handle.cx,
handle.limits,
"kv",
"delete",
&handle.collection,
@@ -271,10 +254,8 @@ fn group_run_before(
value: Option<&serde_json::Value>,
) -> Result<(), Box<EvalAltResult>> {
super::interceptor::run_before(
&handle.interceptors,
handle.self_engine.as_ref(),
&handle.ictx,
&handle.cx,
handle.limits,
"kv",
op,
&handle.collection,

20
crates/executor-core/src/sdk/mod.rs
View File

@@ -60,13 +60,21 @@ pub fn register_all(
limits: Limits,
self_engine: Option<Arc<Engine>>,
) {
kv::register(engine, services, cx.clone(), limits, self_engine.clone());
docs::register(engine, services, cx.clone());
// §9.4 M1: build the interceptor ctx once and thread it into every SDK
// service that has (or will gain) a before/after hook. Only `kv` runs a
// hook today; docs/files/queue/pubsub/http carry it for M7–M11.
let ictx = interceptor::InterceptorCtx {
interceptors: services.interceptors.clone(),
self_engine: self_engine.clone(),
limits,
};
kv::register(engine, services, cx.clone(), ictx.clone());
docs::register(engine, services, cx.clone(), ictx.clone());
dead_letters::register(engine, services, cx.clone());
http::register(engine, services, cx.clone());
files::register(engine, services, cx.clone());
pubsub::register(engine, services, cx.clone());
queue::register(engine, services, cx.clone());
http::register(engine, services, cx.clone(), ictx.clone());
files::register(engine, services, cx.clone(), ictx.clone());
pubsub::register(engine, services, cx.clone(), ictx.clone());
queue::register(engine, services, cx.clone(), ictx);
retry::register(engine, services, cx.clone());
secrets::register(engine, services, cx.clone());
vars::register(engine, services, cx.clone());

14
crates/executor-core/src/sdk/pubsub.rs
View File

@@ -25,8 +25,14 @@ use serde_json::Value as Json;
use tokio::runtime::Handle as TokioHandle;
use super::bridge::{block_on, MAX_JSON_MATERIALIZE_BYTES};
use super::interceptor::InterceptorCtx;
pub(super) fn register(engine: &mut RhaiEngine, services: &Services, cx: Arc<SdkCallCx>) {
pub(super) fn register(
engine: &mut RhaiEngine,
services: &Services,
cx: Arc<SdkCallCx>,
ictx: InterceptorCtx,
) {
let svc = services.pubsub.clone();
let mut module = Module::new();
{
@@ -77,6 +83,7 @@ pub(super) fn register(engine: &mut RhaiEngine, services: &Services, cx: Arc<Sdk
{
let group_svc = services.group_pubsub.clone();
let cx = cx.clone();
let ictx = ictx.clone();
module.set_native_fn(
"shared_topic",
move |name: &str| -> Result<GroupTopicHandle, Box<EvalAltResult>> {
@@ -87,6 +94,7 @@ pub(super) fn register(engine: &mut RhaiEngine, services: &Services, cx: Arc<Sdk
namespace: name.to_string(),
service: group_svc.clone(),
cx: cx.clone(),
ictx: ictx.clone(),
})
},
);
@@ -104,6 +112,10 @@ pub struct GroupTopicHandle {
namespace: String,
service: Arc<dyn picloud_shared::GroupPubsubService>,
cx: Arc<SdkCallCx>,
// §9.4 M1 plumbing: carried so `publish` can run a before/after hook in a
// later milestone (M11). Unused until then.
#[allow(dead_code)]
ictx: InterceptorCtx,
}
fn shared_publish(

14
crates/executor-core/src/sdk/queue.rs
View File

@@ -28,8 +28,14 @@ use serde_json::Value as Json;
use tokio::runtime::Handle as TokioHandle;
use super::bridge::MAX_JSON_MATERIALIZE_BYTES;
use super::interceptor::InterceptorCtx;
pub(super) fn register(engine: &mut RhaiEngine, services: &Services, cx: Arc<SdkCallCx>) {
pub(super) fn register(
engine: &mut RhaiEngine,
services: &Services,
cx: Arc<SdkCallCx>,
ictx: InterceptorCtx,
) {
let svc = services.queue.clone();
let mut module = Module::new();
@@ -100,6 +106,7 @@ pub(super) fn register(engine: &mut RhaiEngine, services: &Services, cx: Arc<Sdk
{
let group_svc = services.group_queue.clone();
let cx = cx.clone();
let ictx = ictx.clone();
module.set_native_fn(
"shared_collection",
move |name: &str| -> Result<GroupQueueHandle, Box<EvalAltResult>> {
@@ -110,6 +117,7 @@ pub(super) fn register(engine: &mut RhaiEngine, services: &Services, cx: Arc<Sdk
collection: name.to_string(),
service: group_svc.clone(),
cx: cx.clone(),
ictx: ictx.clone(),
})
},
);
@@ -126,6 +134,10 @@ pub struct GroupQueueHandle {
collection: String,
service: Arc<dyn GroupQueueService>,
cx: Arc<SdkCallCx>,
// §9.4 M1 plumbing: carried so `enqueue` can run a before/after hook in a
// later milestone (M10). Unused until then.
#[allow(dead_code)]
ictx: InterceptorCtx,
}
fn group_enqueue(