mhome_script_runtime/
lib.rs

1//! Native QuickJS execution, independent of Core/cloud storage and host services.
2//! Each call owns its runtime. The local future must run on a dedicated JS worker,
3//! not a request executor thread: JavaScript runs synchronously between awaits.
4mod value;
5mod worker;
6pub use worker::{Admission, ScriptWorkers, WorkerConfig};
7
8use mhome_script_api::{ScriptError, ScriptValue, ValueLimits, MAX_ARTIFACT_BYTES};
9use rquickjs::{
10    function::Args,
11    loader::{ImportAttributes, Loader, Resolver},
12    module::{Declarations, Exports, ModuleDef},
13    promise::MaybePromise,
14    AsyncContext, AsyncRuntime, Ctx, Exception, Function, Module, Promise, Value,
15};
16use std::{
17    future::Future,
18    pin::Pin,
19    sync::{
20        atomic::{AtomicUsize, Ordering},
21        Arc,
22    },
23    time::Instant,
24};
25pub use tokio_util::sync::CancellationToken;
26
27pub type HostFuture<'a> =
28    Pin<Box<dyn Future<Output = Result<ScriptValue, ScriptError>> + Send + 'a>>;
29
30/// Host implementations own trusted tenant/Space/connection context. They must
31/// allowlist operations, authorize each call, and bound IO before allocating it.
32/// Dropping these futures releases temporary resources; it is not compensation.
33pub trait ScriptHost: Send + Sync + 'static {
34    fn call<'a>(
35        &'a self,
36        operation: &'a str,
37        args: Vec<ScriptValue>,
38        control: InvocationControl,
39    ) -> HostFuture<'a>;
40}
41
42pub struct NoHost;
43impl ScriptHost for NoHost {
44    fn call<'a>(&'a self, _: &'a str, _: Vec<ScriptValue>, _: InvocationControl) -> HostFuture<'a> {
45        Box::pin(async {
46            Err(ScriptError::new(
47                "HOST_UNAVAILABLE",
48                "This script has no host capabilities",
49            ))
50        })
51    }
52}
53
54#[derive(Clone, Debug)]
55pub struct InvocationControl {
56    pub deadline: Instant,
57    pub cancel: CancellationToken,
58    shutdown: CancellationToken,
59}
60
61impl InvocationControl {
62    pub fn new(deadline: Instant, cancel: CancellationToken) -> Self {
63        Self {
64            deadline,
65            cancel,
66            shutdown: CancellationToken::new(),
67        }
68    }
69
70    pub async fn cancelled(&self) {
71        tokio::select! {
72            _ = self.cancel.cancelled() => {},
73            _ = self.shutdown.cancelled() => {},
74        }
75    }
76
77    pub fn check(&self) -> Result<(), ScriptError> {
78        if self.cancel.is_cancelled() || self.shutdown.is_cancelled() {
79            Err(ScriptError::new("CANCELLED", "Script invocation cancelled"))
80        } else if Instant::now() >= self.deadline {
81            Err(ScriptError::new(
82                "DEADLINE_EXCEEDED",
83                "Script invocation timed out",
84            ))
85        } else {
86            Ok(())
87        }
88    }
89}
90
91#[derive(Clone, Copy, Debug)]
92pub struct ScriptLimits {
93    pub memory_bytes: usize,
94    pub stack_bytes: usize,
95    pub values: ValueLimits,
96    pub host_calls: usize,
97    pub concurrent_host_calls: usize,
98}
99
100impl Default for ScriptLimits {
101    fn default() -> Self {
102        Self {
103            memory_bytes: 32 * 1024 * 1024,
104            stack_bytes: 512 * 1024,
105            values: ValueLimits::default(),
106            host_calls: 1024,
107            concurrent_host_calls: 32,
108        }
109    }
110}
111
112struct CancelOnDrop(CancellationToken);
113impl Drop for CancelOnDrop {
114    fn drop(&mut self) {
115        self.0.cancel();
116    }
117}
118
119/// Source must already be resolved against trusted ownership, manifest and hash.
120/// This leaf executor does not grant resource access or fetch artifacts.
121pub async fn invoke(
122    source: &str,
123    args: Vec<ScriptValue>,
124    host: Arc<dyn ScriptHost>,
125    control: InvocationControl,
126    limits: ScriptLimits,
127) -> Result<ScriptValue, ScriptError> {
128    control.check()?;
129    if source.is_empty() || source.len() > MAX_ARTIFACT_BYTES {
130        return Err(ScriptError::new(
131            "INVALID_SCRIPT",
132            "JavaScript source size is out of bounds",
133        ));
134    }
135    let args = ScriptValue::List(args);
136    args.validate(limits.values)
137        .map_err(|e| ScriptError::new("INVALID_ARGUMENT", e.to_string()))?;
138    let control = InvocationControl {
139        cancel: control.cancel.child_token(),
140        ..control
141    };
142    let _cancel_on_drop = CancelOnDrop(control.cancel.clone());
143    let runtime = AsyncRuntime::new().map_err(engine_error)?;
144    runtime.set_memory_limit(limits.memory_bytes).await;
145    runtime.set_max_stack_size(limits.stack_bytes).await;
146    let interrupt = control.clone();
147    runtime
148        .set_interrupt_handler(Some(Box::new(move || interrupt.check().is_err())))
149        .await;
150    runtime.set_loader(HostResolver, HostLoader).await;
151    let rejected = Arc::new(AtomicUsize::new(0));
152    let rejection_count = rejected.clone();
153    runtime
154        .set_host_promise_rejection_tracker(Some(Box::new(move |_, _, _, handled| {
155            if handled {
156                let _ = rejection_count.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |n| {
157                    Some(n.saturating_sub(1))
158                });
159            } else {
160                rejection_count.fetch_add(1, Ordering::Relaxed);
161            }
162        })))
163        .await;
164    let context = AsyncContext::full(&runtime).await.map_err(engine_error)?;
165    let active = Arc::new(AtomicUsize::new(0));
166    let call = context.async_with(async |ctx| {
167        let result = async {
168            install_host(&ctx, host, control.clone(), limits, active.clone())?;
169            let (module, ready) = Module::declare(ctx.clone(), "entry.mjs", source)?.eval()?;
170            ready.into_future::<()>().await?;
171            let entry: Function = module.get("default")?;
172            let ScriptValue::List(values) = &args else {
173                unreachable!()
174            };
175            let mut arguments = Args::new(ctx.clone(), values.len());
176            for arg in values {
177                arguments.push_arg(value::to_js(&ctx, arg)?)?;
178            }
179            let result: MaybePromise = arguments.apply(&entry)?;
180            let output = result.into_future::<Value>().await?;
181            // Finish queued JS microtasks, but never drive newly detached host IO.
182            // Interrupts and this check also bound a chain of tiny Promise jobs.
183            while ctx.execute_pending_job() {
184                control.check().map_err(|e| throw_host(&ctx, e))?;
185            }
186            value::from_js(output, limits.values)
187        }
188        .await;
189        result.map_err(|error| exception_error(&ctx, error))
190    });
191    let result = tokio::select! {
192        biased;
193        _ = control.cancelled() => Err(ScriptError::new("CANCELLED", "Script invocation cancelled")),
194        _ = tokio::time::sleep_until(control.deadline.into()) => Err(ScriptError::new("DEADLINE_EXCEEDED", "Script invocation timed out")),
195        result = call => result,
196    };
197    // Interrupts report JS exceptions; expose the actual request cancellation reason.
198    control.check()?;
199    if result.is_ok() && active.load(Ordering::Acquire) != 0 {
200        return Err(ScriptError::new(
201            "UNAWAITED_HOST_CALL",
202            "Script returned before its host calls completed",
203        ));
204    }
205    if result.is_ok() && rejected.load(Ordering::Acquire) != 0 {
206        return Err(ScriptError::new(
207            "UNHANDLED_REJECTION",
208            "Script left an unhandled Promise rejection",
209        ));
210    }
211    result
212    // Dropping context/runtime destroys their pending Promise/host futures. Never
213    // re-enter failed guest code for cleanup or tear down host-owned subscriptions.
214}
215
216fn engine_error(error: rquickjs::Error) -> ScriptError {
217    ScriptError::new("SCRIPT_ERROR", bounded(&error.to_string(), 4096))
218}
219
220fn bounded(value: &str, max: usize) -> String {
221    let mut end = value.len().min(max);
222    while !value.is_char_boundary(end) {
223        end -= 1;
224    }
225    value[..end].to_owned()
226}
227
228fn exception_error(ctx: &Ctx<'_>, error: rquickjs::Error) -> ScriptError {
229    if !error.is_exception() {
230        return engine_error(error);
231    }
232    let exception = ctx.catch();
233    if let Some(object) = exception.as_object() {
234        let code = object
235            .get::<_, String>("code")
236            .unwrap_or_else(|_| "SCRIPT_ERROR".into());
237        let message = object
238            .get::<_, String>("message")
239            .unwrap_or_else(|_| "Script failed".into());
240        ScriptError::new(bounded(&code, 128), bounded(&message, 4096))
241    } else if let Some(message) = exception.as_string().and_then(|v| v.to_string().ok()) {
242        ScriptError::new("SCRIPT_ERROR", bounded(&message, 4096))
243    } else {
244        ScriptError::new("SCRIPT_ERROR", "Script threw a non-error value")
245    }
246}
247
248fn throw_host(ctx: &Ctx<'_>, error: ScriptError) -> rquickjs::Error {
249    match Exception::from_message(ctx.clone(), &bounded(&error.message, 4096)) {
250        Ok(exception) => {
251            if let Err(error) = exception.as_object().set("code", bounded(&error.code, 128)) {
252                return error;
253            }
254            exception.throw()
255        }
256        Err(error) => error,
257    }
258}
259
260struct ActiveCall(Arc<AtomicUsize>);
261impl Drop for ActiveCall {
262    fn drop(&mut self) {
263        self.0.fetch_sub(1, Ordering::AcqRel);
264    }
265}
266
267fn install_host<'js>(
268    ctx: &Ctx<'js>,
269    host: Arc<dyn ScriptHost>,
270    control: InvocationControl,
271    limits: ScriptLimits,
272    active: Arc<AtomicUsize>,
273) -> rquickjs::Result<()> {
274    let calls = Arc::new(AtomicUsize::new(0));
275    let function = Function::new(
276        ctx.clone(),
277        move |ctx: Ctx<'js>, operation: String, args: Value<'js>| {
278            control.check().map_err(|e| throw_host(&ctx, e))?;
279            if operation.is_empty() || operation.len() > 128 {
280                return Err(Exception::throw_type(&ctx, "invalid host operation"));
281            }
282            let ScriptValue::List(args) = value::from_js(args, limits.values)? else {
283                return Err(Exception::throw_type(
284                    &ctx,
285                    "host arguments must be an array",
286                ));
287            };
288            if calls.fetch_add(1, Ordering::Relaxed) >= limits.host_calls {
289                return Err(Exception::throw_range(
290                    &ctx,
291                    "script host call limit exceeded",
292                ));
293            }
294            if active
295                .fetch_update(Ordering::AcqRel, Ordering::Acquire, |n| {
296                    (n < limits.concurrent_host_calls).then_some(n + 1)
297                })
298                .is_err()
299            {
300                return Err(Exception::throw_range(
301                    &ctx,
302                    "script concurrent host call limit exceeded",
303                ));
304            }
305            let guard = ActiveCall(active.clone());
306            let host = host.clone();
307            let control = control.clone();
308            let future_ctx = ctx.clone();
309            Promise::wrap_future(&ctx, async move {
310                let _guard = guard;
311                control.check().map_err(|e| throw_host(&future_ctx, e))?;
312                let result = tokio::select! {
313                biased;
314                _ = control.cancelled() => Err(ScriptError::new("CANCELLED", "Host call cancelled")),
315                _ = tokio::time::sleep_until(control.deadline.into()) => Err(ScriptError::new("DEADLINE_EXCEEDED", "Host call timed out")),
316                result = host.call(&operation, args, control.clone()) => result,
317            }.map_err(|e| throw_host(&future_ctx, e))?;
318                control.check().map_err(|e| throw_host(&future_ctx, e))?;
319                result
320                    .validate(limits.values)
321                    .map_err(|e| Exception::throw_range(&future_ctx, &e.to_string()))?;
322                value::to_js(&future_ctx, &result)
323            })
324        },
325    )?;
326    // Function has a safe JsLifetime implementation. The native module consumes
327    // this private value during evaluation; it is never a guest-writable global.
328    ctx.store_userdata(function)
329        .map_err(|_| Exception::throw_internal(ctx, "host bindings already borrowed"))?;
330    Ok(())
331}
332
333const HOST_MODULE: &str = "@meowlink/script/host";
334struct HostResolver;
335impl Resolver for HostResolver {
336    fn resolve<'js>(
337        &mut self,
338        _: &Ctx<'js>,
339        base: &str,
340        name: &str,
341        attributes: Option<ImportAttributes<'js>>,
342    ) -> rquickjs::Result<String> {
343        if name == HOST_MODULE && attributes.is_none() {
344            Ok(name.into())
345        } else {
346            Err(rquickjs::Error::new_resolving_message(
347                base,
348                name,
349                "only bundled code and platform host imports are supported",
350            ))
351        }
352    }
353}
354struct HostLoader;
355impl Loader for HostLoader {
356    fn load<'js>(
357        &mut self,
358        ctx: &Ctx<'js>,
359        name: &str,
360        _: Option<ImportAttributes<'js>>,
361    ) -> rquickjs::Result<Module<'js>> {
362        if name == HOST_MODULE {
363            Module::declare_def::<HostModule, _>(ctx.clone(), name)
364        } else {
365            Err(rquickjs::Error::new_loading(name))
366        }
367    }
368}
369struct HostModule;
370impl ModuleDef for HostModule {
371    fn declare<'js>(declarations: &Declarations<'js>) -> rquickjs::Result<()> {
372        declarations.declare("call")?;
373        Ok(())
374    }
375    fn evaluate<'js>(ctx: &Ctx<'js>, exports: &Exports<'js>) -> rquickjs::Result<()> {
376        let function = ctx
377            .remove_userdata::<Function<'js>>()
378            .map_err(|_| Exception::throw_internal(ctx, "host bindings borrowed"))?
379            .ok_or_else(|| Exception::throw_internal(ctx, "host bindings unavailable"))?;
380        exports.export("call", *function)?;
381        Ok(())
382    }
383}
mhome_script_runtime/lib.rs

mhome_script_runtime/
lib.rs
