apiplant-js 0.6.1

//! One V8 isolate on one thread, and the channels that let the rest of the
//! process talk to it.
//!
//! A V8 isolate belongs to the thread that created it, so a JavaScript function
//! cannot simply be called on whatever worker the HTTP server happens to be
//! using. Instead each isolate gets a thread of its own and receives [`Job`]s;
//! the caller blocks until the answer comes back, which makes an invocation look
//! synchronous from the outside — exactly like the `.so` path next to it.
//!
//! ## Who runs the host calls
//!
//! When the function asks the host for something — a query, its config — the op
//! does **not** run it on the isolate's thread. It sends a [`Message::Host`] back
//! to the caller and blocks. The caller is a `spawn_blocking` worker that already
//! holds the [`HostApi`](apiplant_abi::HostApi) and is allowed to block on the
//! async runtime; the isolate's thread is neither. So the request travels back to
//! the one thread that can serve it, and the reply travels forward.
//!
//! That inversion is the whole design: the isolate thread only ever runs
//! JavaScript, and the ops are a mailbox.

use std::cell::RefCell;
use std::rc::Rc;
use std::time::Duration;

use crossbeam_channel::{bounded, Receiver, Sender};
use deno_core::{
    extension, op2, ExtensionFileSource, JsRuntime, OpState, PollEventLoopOptions, RuntimeOptions,
};

/// How long one invocation may run before the isolate is terminated.
///
/// A JavaScript function is not preemptible: `while (true) {}` would hold its
/// thread until the process ended, and a pool of them can be exhausted by one
/// bad deployment. V8 can interrupt a running isolate from another thread, which
/// is what the watchdog below does.
fn timeout() -> Duration {
    let ms = std::env::var("APIPLANT_JS_TIMEOUT_MS")
        .ok()
        .and_then(|v| v.parse::<u64>().ok())
        .unwrap_or(30_000);
    Duration::from_millis(ms)
}

/// One call into a JavaScript function.
pub(crate) struct Job {
    /// The manifest name, which is also the export to call.
    pub name: String,
    /// The request body as JSON.
    pub input: String,
    /// Where the isolate sends host requests and, finally, the result.
    pub replies: Sender<Message>,
}

/// What the isolate's thread sends back over a job's channel.
pub(crate) enum Message {
    /// The function asked the host for something and is blocked until answered.
    Host {
        kind: String,
        payload: String,
        answer: Sender<String>,
    },
    /// The call finished. Always the last message of a job.
    Done(Result<String, String>),
}

/// The isolate-side end of the current job, read by the op.
type Current = Rc<RefCell<Option<Sender<Message>>>>;

#[op2]
#[string]
fn op_apiplant_host(state: &mut OpState, #[string] kind: &str, #[string] payload: &str) -> String {
    let current = state.borrow::<Current>().clone();
    let replies = current.borrow().clone();
    let Some(replies) = replies else {
        // Only reachable if user code stashed `ctx` and called it after the
        // invocation returned — from a timer, say. There is no host to ask.
        return r#"{"error":"no invocation in progress"}"#.to_string();
    };

    let (answer, wait) = bounded(1);
    let sent = replies.send(Message::Host {
        kind: kind.to_string(),
        payload: payload.to_string(),
        answer,
    });
    if sent.is_err() {
        return r#"{"error":"the host stopped listening"}"#.to_string();
    }
    wait.recv()
        .unwrap_or_else(|_| r#"{"error":"the host stopped listening"}"#.to_string())
}

/// The specifier the bootstrap is registered and entered under.
const BOOTSTRAP: &str = "ext:apiplant_js/bootstrap.js";

/// The bootstrap itself, compiled into the binary.
///
/// It lives in `assets/` and arrives here through `include_str!` for one
/// reason: `extension!`'s `esm = [dir …]` form does *not* embed the file. It
/// expands to `ExtensionFileSource::loaded_during_snapshot`, which stores the
/// build machine's absolute `CARGO_MANIFEST_DIR` path and reads it off disk when
/// an isolate starts. That works in a checkout and nowhere else: a released
/// binary looks for the CI runner's source tree, does not find it, and every
/// TypeScript function fails to load while the rest of the server is fine. So
/// the source is embedded here and installed over `esm_files` below.
const BOOTSTRAP_SOURCE: &str = include_str!("../assets/bootstrap.js");

extension!(
    apiplant_js,
    ops = [op_apiplant_host],
    esm_entry_point = BOOTSTRAP,
    options = { current: Current },
    state = |state, options| state.put::<Current>(options.current),
);

/// The extension, with the bootstrap supplied as source rather than a path.
fn extension(current: Current) -> deno_core::Extension {
    let mut extension = apiplant_js::init(current);
    extension.esm_files = std::borrow::Cow::Owned(vec![ExtensionFileSource::new_computed(
        BOOTSTRAP,
        BOOTSTRAP_SOURCE.into(),
    )]);
    extension
}

/// Start an isolate on its own thread with `code` already evaluated.
///
/// Returns the manifest the module declared, plus the channel jobs go down.
/// Failing to compile or evaluate the module fails here rather than at the first
/// request, so a broken function library is reported at boot like a broken `.so`.
pub(crate) fn spawn(
    label: String,
    code: String,
    jobs: Receiver<Job>,
) -> Result<Option<String>, String> {
    let (ready, wait) = bounded::<Result<Option<String>, String>>(1);

    std::thread::Builder::new()
        .name(format!("apiplant-js:{label}"))
        .spawn(move || run(label, code, jobs, ready))
        .map_err(|e| format!("cannot start a JavaScript worker thread: {e}"))?;

    wait.recv()
        .map_err(|_| "the JavaScript worker died during startup".to_string())?
}

/// The isolate thread: build the runtime, evaluate the module, serve jobs.
fn run(
    label: String,
    code: String,
    jobs: Receiver<Job>,
    ready: Sender<Result<Option<String>, String>>,
) {
    let current: Current = Rc::new(RefCell::new(None));
    let mut runtime = JsRuntime::new(RuntimeOptions {
        extensions: vec![extension(current.clone())],
        // Serves `import … from "apiplant"` and refuses everything else.
        module_loader: Some(crate::module::Loader::shared()),
        ..Default::default()
    });

    // The isolate's own event loop needs an async context to be driven in; it is
    // current-thread and single-purpose, and never blocks on anything but V8,
    // because host work happens on the *caller's* thread (see the module docs).
    let Ok(local) = tokio::runtime::Builder::new_current_thread()
        .enable_time()
        .build()
    else {
        let _ = ready.send(Err("cannot start the JavaScript event loop".into()));
        return;
    };

    let watchdog = Watchdog::spawn(runtime.v8_isolate().thread_safe_handle());

    let manifest = local.block_on(evaluate(&mut runtime, &label, code));
    let failed = manifest.is_err();
    let _ = ready.send(manifest);
    if failed {
        return;
    }

    // `__apiplantInvoke` is fetched once: it is the only entry point, and looking
    // it up per call would mean a handle scope per call for nothing.
    let entry = match global_function(&mut runtime, "__apiplantInvoke") {
        Ok(f) => f,
        Err(e) => {
            tracing::error!(library = %label, error = %e, "javascript bootstrap is broken");
            return;
        }
    };

    // A closed job channel means the registry is gone: the process is shutting
    // down, so the isolate goes with it.
    while let Ok(job) = jobs.recv() {
        *current.borrow_mut() = Some(job.replies.clone());
        let guard = watchdog.watching();
        let result = local.block_on(invoke(&mut runtime, &entry, &job.name, &job.input));
        drop(guard);
        *current.borrow_mut() = None;

        // A terminated isolate stays terminated until told otherwise; without
        // this the worker would reject every later request too.
        runtime.v8_isolate().cancel_terminate_execution();

        let _ = job.replies.send(Message::Done(result));
    }
}

/// Load and evaluate the module, then read its manifest.
async fn evaluate(
    runtime: &mut JsRuntime,
    label: &str,
    code: String,
) -> Result<Option<String>, String> {
    // The specifier is only ever seen in stack traces, so it names the library.
    let url = deno_core::resolve_url(&format!("file:///{label}.js"))
        .map_err(|e| format!("cannot name the module: {e}"))?;

    let id = runtime
        .load_main_es_module_from_code(&url, code)
        .await
        .map_err(|e| format!("cannot compile the module: {e}"))?;
    let evaluated = runtime.mod_evaluate(id);
    runtime
        .run_event_loop(PollEventLoopOptions::default())
        .await
        .map_err(|e| format!("module failed while evaluating: {e}"))?;
    evaluated
        .await
        .map_err(|e| format!("module failed while evaluating: {e}"))?;

    // Hand the namespace to the bootstrap, which is what dispatches into it.
    let namespace = runtime
        .get_module_namespace(id)
        .map_err(|e| format!("cannot read the module's exports: {e}"))?;
    {
        deno_core::scope!(scope, runtime);
        let namespace = deno_core::v8::Local::new(scope, namespace);
        let global = scope.get_current_context().global(scope);
        let key = deno_core::v8::String::new(scope, "__apiplantModule")
            .ok_or("out of memory naming the module")?;
        global.set(scope, key.into(), namespace.into());
    }

    let manifest = global_function(runtime, "__apiplantManifest")?;
    let manifest = invoke_json(runtime, &manifest, &[]).await?;
    Ok(match manifest.as_str() {
        "" | "null" => None,
        json => Some(json.to_string()),
    })
}

/// Call `__apiplantInvoke(name, input)` and unwrap what it resolves to.
async fn invoke(
    runtime: &mut JsRuntime,
    entry: &deno_core::v8::Global<deno_core::v8::Function>,
    name: &str,
    input: &str,
) -> Result<String, String> {
    let args = {
        deno_core::scope!(scope, runtime);
        let name: deno_core::v8::Local<deno_core::v8::Value> =
            deno_core::v8::String::new(scope, name)
                .ok_or("out of memory")?
                .into();
        let input: deno_core::v8::Local<deno_core::v8::Value> =
            deno_core::v8::String::new(scope, input)
                .ok_or("out of memory")?
                .into();
        [
            deno_core::v8::Global::new(scope, name),
            deno_core::v8::Global::new(scope, input),
        ]
    };

    // The bootstrap resolves rather than rejects, so a failure here is the
    // isolate itself failing: a timeout, an out-of-memory, a top-level throw
    // from a timer. All of those are the function's fault, never the caller's.
    let reply = invoke_json(runtime, entry, &args).await.map_err(|e| {
        format!(
            "{}javascript function `{name}` failed: {e}",
            apiplant_abi::INTERNAL_ERROR_PREFIX
        )
    })?;

    let reply: serde_json::Value = serde_json::from_str(&reply).map_err(|e| {
        format!(
            "{}invoke returned invalid JSON: {e}",
            apiplant_abi::INTERNAL_ERROR_PREFIX
        )
    })?;

    if let Some(error) = reply.get("error").and_then(|e| e.as_str()) {
        // `request: true` is a 400 and goes back bare; anything else is a 500,
        // which the host recognises by the prefix.
        let caller_fault = reply.get("request").and_then(|r| r.as_bool()) == Some(true);
        return Err(if caller_fault {
            error.to_string()
        } else {
            format!("{}{error}", apiplant_abi::INTERNAL_ERROR_PREFIX)
        });
    }
    Ok(match reply.get("ok") {
        Some(value) => value.to_string(),
        None => "null".to_string(),
    })
}

/// Call a JavaScript function that returns a string (or a promise of one),
/// driving the event loop until it settles.
async fn invoke_json(
    runtime: &mut JsRuntime,
    function: &deno_core::v8::Global<deno_core::v8::Function>,
    args: &[deno_core::v8::Global<deno_core::v8::Value>],
) -> Result<String, String> {
    let call = runtime.call_with_args(function, args);
    let value = runtime
        .with_event_loop_promise(call, PollEventLoopOptions::default())
        .await
        .map_err(|e| e.to_string())?;

    deno_core::scope!(scope, runtime);
    let value = deno_core::v8::Local::new(scope, value);
    if value.is_null_or_undefined() {
        return Ok(String::new());
    }
    Ok(value.to_rust_string_lossy(scope))
}

/// Fetch a function off the global object, by name.
fn global_function(
    runtime: &mut JsRuntime,
    name: &str,
) -> Result<deno_core::v8::Global<deno_core::v8::Function>, String> {
    deno_core::scope!(scope, runtime);
    let global = scope.get_current_context().global(scope);
    let key = deno_core::v8::String::new(scope, name).ok_or("out of memory")?;
    let value = global
        .get(scope, key.into())
        .ok_or_else(|| format!("`{name}` is missing from the isolate"))?;
    let function: deno_core::v8::Local<deno_core::v8::Function> = value
        .try_into()
        .map_err(|_| format!("`{name}` is not a function"))?;
    Ok(deno_core::v8::Global::new(scope, function))
}

/// Terminates an isolate that overstays its [`timeout`].
///
/// Lives on its own thread because the isolate's thread is, by definition, busy
/// running the code that needs interrupting.
struct Watchdog {
    signals: Sender<Signal>,
    timeout: Duration,
}

enum Signal {
    Begin(Duration),
    End,
}

impl Watchdog {
    fn spawn(handle: deno_core::v8::IsolateHandle) -> Watchdog {
        let (signals, incoming) = bounded::<Signal>(1);
        std::thread::Builder::new()
            .name("apiplant-js:watchdog".into())
            .spawn(move || {
                while let Ok(Signal::Begin(limit)) = incoming.recv() {
                    // Either the call ends in time, or V8 is interrupted and the
                    // `End` that follows the failure is absorbed here.
                    if incoming.recv_timeout(limit).is_err() {
                        handle.terminate_execution();
                        if incoming.recv().is_err() {
                            return;
                        }
                    }
                }
            })
            .ok();
        Watchdog {
            signals,
            timeout: timeout(),
        }
    }

    /// Arm the watchdog for one call; disarmed when the guard drops.
    fn watching(&self) -> WatchGuard<'_> {
        let _ = self.signals.send(Signal::Begin(self.timeout));
        WatchGuard { watchdog: self }
    }
}

struct WatchGuard<'a> {
    watchdog: &'a Watchdog,
}

impl Drop for WatchGuard<'_> {
    fn drop(&mut self) {
        let _ = self.watchdog.signals.send(Signal::End);
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    /// The bootstrap must travel *inside* the binary, not as a path to it.
    ///
    /// This is a regression test for a released binary that loaded no
    /// TypeScript function at all: `extension!`'s `esm = [dir …]` form records
    /// the build machine's absolute path and reads it when an isolate starts,
    /// so every function failed on any machine that was not the CI runner. A
    /// checkout cannot notice — the path is right there — which is why the
    /// assertion is on the extension rather than on a working isolate.
    #[test]
    fn bootstrap_is_embedded_not_a_path() {
        let extension = extension(Rc::new(RefCell::new(None)));

        let bootstrap = extension
            .esm_files
            .iter()
            .find(|file| file.specifier == BOOTSTRAP)
            .expect("the extension must carry the bootstrap");

        assert!(
            bootstrap.is_runtime_loadable(),
            "{BOOTSTRAP} is a build-machine path, so a released binary cannot load it",
        );
        for file in extension.esm_files.iter() {
            assert!(file.is_runtime_loadable(), "{} is a path", file.specifier);
        }
    }

    /// The entry point has to name a module the extension actually supplies;
    /// getting this wrong fails only at isolate startup, in the same silent way.
    #[test]
    fn bootstrap_is_the_entry_point() {
        let extension = extension(Rc::new(RefCell::new(None)));
        assert_eq!(extension.esm_entry_point, Some(BOOTSTRAP));
    }
}