diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 4f40be8..4ef4eb5 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -35,12 +35,20 @@ jobs: deno-version: ${{ env.DENO_VERSION }} - uses: taiki-e/install-action@v2 with: - tool: just@${{ env.JUST_VERSION }},wasm-tools@${{ env.WASM_TOOLS_VERSION }} + tool: just@${{ env.JUST_VERSION }},wasm-tools@${{ env.WASM_TOOLS_VERSION }},cargo-binstall - uses: Swatinem/rust-cache@v2 with: workspaces: | . guests/web-sys + host + # The desktop crate (host/desktop) links webkit2gtk/gtk3; its + # WebDriver smoke test also needs WebKitWebDriver + Xvfb (no + # display on this runner). + - run: | + sudo apt-get update + sudo apt-get install -y libwebkit2gtk-4.1-dev libgtk-3-dev librsvg2-dev \ + libsoup-3.0-dev libjavascriptcoregtk-4.1-dev webkit2gtk-driver xvfb - run: just check - run: just test - run: just components @@ -52,3 +60,6 @@ jobs: - run: npx -y playwright@1.58 install --with-deps chromium - run: just e2e - run: just bench-wire + # Prebuilt: `cargo install` would compile it on every run. + - run: cargo binstall -y tauri-driver + - run: just desktop-smoke diff --git a/.gitignore b/.gitignore index 0960c5d..0ee1181 100644 --- a/.gitignore +++ b/.gitignore @@ -8,3 +8,8 @@ test-results/ playwright-report/ bench/results/ bench/tachometer.json +# Tauri's generated capability schemas (host/desktop/gen/schemas/...). +gen/ +# Regenerated by tauri-build (AppManifest::commands) on every build. +host/desktop/permissions/autogenerated/ + diff --git a/Cargo.toml b/Cargo.toml index eb281a9..d3044c6 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -11,7 +11,10 @@ members = [ # js-sys, web-sys and wasm-bindgen-futures with the fake-DOM shims, and a # patch applies workspace-wide. Dioxus's graph pulls the real wasm-bindgen # (via subsecond) as dead code, which must not see the shims. -exclude = ["guests/web-sys"] +# +# The wasmtime host is its own workspace because this one is built for +# wasm32-wasip2 and wasmtime cannot be. +exclude = ["guests/web-sys", "host"] [workspace.package] edition = "2021" diff --git a/README.md b/README.md index b25f5d7..76c1332 100644 --- a/README.md +++ b/README.md @@ -37,7 +37,10 @@ latter is its own cargo workspace because it `[patch]`es `wasm-bindgen` with a fake whose `#[wasm_bindgen]` macro lets the real `web-sys` drive a Rust shadow DOM — Dominator's TodoMVC runs unmodified), `receiver/` (TypeScript receiver + polyengine host glue), `web/` (demo -site and browser test). `just --list` for the build and test recipes. +site and browser test), `host/` (a wasmtime host for producers and a Tauri +desktop app that renders one through the same receiver in its webview — +its own cargo workspace, since it is native). `just --list` for the build +and test recipes. The immediate predecessor is [polyengine-dioxus](https://github.com/lannbot/polyengine-dioxus), a diff --git a/crates/stream-dom-guest/src/lib.rs b/crates/stream-dom-guest/src/lib.rs index d0de16f..a03f479 100644 --- a/crates/stream-dom-guest/src/lib.rs +++ b/crates/stream-dom-guest/src/lib.rs @@ -187,7 +187,13 @@ impl Batch { /// Set an attribute to an opaque asset handle (proto/stream-dom.proto /// `SetAttribute.asset`) rather than a text value. - pub fn set_attribute_asset(&mut self, id: NodeId, name: StrRef, ns: Option, handle: &[u8]) { + pub fn set_attribute_asset( + &mut self, + id: NodeId, + name: StrRef, + ns: Option, + handle: &[u8], + ) { self.push(proto::frame::Op::SetAttribute(proto::SetAttribute { id, name, diff --git a/crates/stream-dom-proto/build.rs b/crates/stream-dom-proto/build.rs index c2759be..655e487 100644 --- a/crates/stream-dom-proto/build.rs +++ b/crates/stream-dom-proto/build.rs @@ -21,16 +21,18 @@ fn main() { .compile_fds(file_descriptor_set) .expect("prost-build: compile failed"); - let stream_dom_proto = - std::fs::read_to_string(format!("{proto_dir}/stream-dom.proto")).expect("read stream-dom.proto"); + let stream_dom_proto = std::fs::read_to_string(format!("{proto_dir}/stream-dom.proto")) + .expect("read stream-dom.proto"); let mut matches = stream_dom_proto .lines() .filter_map(|line| line.strip_prefix("// PROTOCOL VERSION: ")); - let version = matches - .next() - .unwrap_or_else(|| panic!("no `// PROTOCOL VERSION: ` line found in {proto_dir}/stream-dom.proto")); + let version = matches.next().unwrap_or_else(|| { + panic!("no `// PROTOCOL VERSION: ` line found in {proto_dir}/stream-dom.proto") + }); if matches.next().is_some() { - panic!("more than one `// PROTOCOL VERSION: ` line found in {proto_dir}/stream-dom.proto"); + panic!( + "more than one `// PROTOCOL VERSION: ` line found in {proto_dir}/stream-dom.proto" + ); } let version: u32 = version .parse() diff --git a/deno.json b/deno.json index 66bff1a..da64242 100644 --- a/deno.json +++ b/deno.json @@ -1,5 +1,5 @@ { - "workspace": ["./receiver", "./web"], + "workspace": ["./receiver", "./web", "./host/desktop"], "imports": { "@polyengine/runtime/embedder": "jsr:@polyengine/runtime@^0.6.2/embedder", "@polyengine/runtime/shim": "jsr:@polyengine/runtime@^0.6.2/shim", @@ -15,14 +15,16 @@ "@std/fs": "jsr:@std/fs@^1", "@std/http/file-server": "jsr:@std/http@^1/file-server", "playwright": "npm:playwright@^1.58", - "linkedom": "npm:linkedom@^0.18" + "linkedom": "npm:linkedom@^0.18", + "@tauri-apps/api": "npm:@tauri-apps/api@^2" }, "compilerOptions": { "lib": ["deno.ns", "dom", "dom.iterable", "esnext"], "strict": true }, "tasks": { - "check": "deno check receiver/src receiver/tests web/build.ts web/translate.ts web/entry.ts web/bench.ts web/e2e bench/tachometer.ts bench/run.ts bench/wire.ts", + "check": "deno check receiver/src receiver/tests web/build.ts web/translate.ts web/entry.ts web/bench.ts web/e2e bench/tachometer.ts bench/run.ts bench/wire.ts host/desktop/ui host/desktop/build.ts host/desktop/e2e", "test": "deno test --allow-read=. receiver/tests/" } } + diff --git a/deno.lock b/deno.lock index 7ff2d65..1da1689 100644 --- a/deno.lock +++ b/deno.lock @@ -25,6 +25,7 @@ "jsr:@std/path@^1.1.6": "1.1.6", "jsr:@std/streams@^1.1.2": "1.1.2", "npm:@remote-dom/core@^1.11.1": "1.11.1", + "npm:@tauri-apps/api@2": "2.11.1", "npm:linkedom@0.18": "0.18.13", "npm:playwright@1.58": "1.58.2", "npm:playwright@^1.58.0": "1.58.2" @@ -120,6 +121,9 @@ "@remote-dom/polyfill@1.5.1": { "integrity": "sha512-eaWdIVKZpNfbqspKkRQLVxiFv/7vIw8u0FVA5oy52YANFbO/WVT0GU+PQmRt/QUSijaB36HBAqx7stjo8HGpVQ==" }, + "@tauri-apps/api@2.11.1": { + "integrity": "sha512-M2FPuYND2m+wh5hfW9ZpSdxMPdEJovPBWwoHJmwUpysTYNHaOkVFN419m/K0LIgjb/7KU2vBgsUepJWugQCvAA==" + }, "boolbase@2.0.0": { "integrity": "sha512-DkVaaQHymRhpYEYo9x1oo7Q7B0Y6KJUsjm3c9eTyFDby4MHLBTwZ6ZDWBel5zrYxj1WsZgC5oLpiz+93MluXeA==" }, @@ -263,6 +267,7 @@ "jsr:@std/http@1", "jsr:@std/path@1", "npm:@remote-dom/core@^1.11.1", + "npm:@tauri-apps/api@2", "npm:linkedom@0.18", "npm:playwright@^1.58.0" ] diff --git a/docs/design.md b/docs/design.md index 66b14a6..e9e7215 100644 --- a/docs/design.md +++ b/docs/design.md @@ -705,13 +705,19 @@ Two gaps the shims cannot close: Design to the weakest transport; the others buy latency, not semantics. -| | In-process (component) | Worker / iframe | Network | -|---|---|---|---| -| Guest blocking reads | yes | yes (host parks the guest across `postMessage`) | yes (slow) | -| Non-`async` export honored inside a DOM listener | yes | no (proxy) | no | -| Imperative `preventDefault` (option A) | yes | no | no | -| Backpressure | rendezvous | rendezvous (host paces reads) | rendezvous (host paces reads by remote ack) | -| Reconnect / resync | n/a | n/a | needs `reset` + full snapshot | +| | In-process (component) | Worker / iframe | Native host ↔ webview | Network | +|---|---|---|---|---| +| Guest blocking reads | yes | yes (host parks the guest across `postMessage`) | yes (host parks the guest across IPC) | yes (slow) | +| Non-`async` export honored inside a DOM listener | yes | no (proxy) | no | no | +| Imperative `preventDefault` (option A) | yes | no | no | no | +| Backpressure | rendezvous | rendezvous (host paces reads) | rendezvous (webview pulls; its next pull acks the last chunk) | rendezvous (host paces reads by remote ack) | +| Reconnect / resync | n/a | n/a | n/a (a dead producer is torn down, not resumed) | needs `reset` + full snapshot | + +The native-host column is the worker tier with the producer in a wasmtime +process instead of a worker: same semantics, and the wasm sandbox gains +what a worker cannot offer — hard memory and CPU limits per producer and a +capability surface (WASI imports) the host chooses. Built in `host/`; see +"Spike". Because the stream is unbuffered, backpressure is uniform: the host decides when to issue the next read and the guest sees identical semantics @@ -811,6 +817,58 @@ structured op fuzzer and a byte-mutation fuzzer over the fixture stream, each asserting the mount root and its siblings are untouched after any rejection. The remote-dom backend is not hardened. +**Trust inversion, and the first concrete policy.** In the browser spike +producer and receiver are equally unprivileged. In a desktop host they are +not: the webview holds IPC capabilities into the native process, so a +DOM-level escape from the producer — a ` + + diff --git a/host/desktop/ui/main.ts b/host/desktop/ui/main.ts new file mode 100644 index 0000000..07236f1 --- /dev/null +++ b/host/desktop/ui/main.ts @@ -0,0 +1,178 @@ +// Browser entry for the desktop app (dispatch: track "host/desktop"). +// Bundled with `deno bundle --platform browser` (build.ts). No fetch/env of +// its own component + protocol version negotiation like web/entry.ts: the +// desktop shell hosts exactly one producer, named by the Rust side's +// `spawn_producer` resource lookup. + +import { Channel, invoke } from "@tauri-apps/api/core"; +import { createDriver, desktopPolicy } from "@polymorph/stream-dom-receiver"; +import type { ProducerEventTarget } from "@polymorph/stream-dom-receiver"; + +// -- Rust -> JS messages on the per-producer Channel ------------------------- +// +// Mirrors `host/desktop/src/bridge.rs`'s `HostMessage` (serde tagged enum, +// `#[serde(rename_all = "kebab-case")]` on both the outer tag and the +// `QueryKind` variants). + +type QueryKind = "client-rect" | "scroll-offset" | "scroll-size" | "focus"; + +type HostMessage = + | { + type: "query"; + id: number; + kind: QueryKind; + target: number; + focus?: boolean; + } + | { type: "closed"; error?: string }; + +interface DesktopStreamDom { + mounted: boolean; + ready: Promise; + stats: { batches: number; frames: number; bytes: number }; + latency: { + reads: number; + lastReadMs: number; + maxPushMs: number; + totalPushMs: number; + }; +} + +declare global { + var __streamDomDesktop: DesktopStreamDom | undefined; +} + +function showError(err: unknown): void { + console.error(err); + const pre = document.getElementById("error"); + if (pre) { + pre.textContent = err instanceof Error + ? `${err.name}: ${err.message}` + : String(err); + } +} + +function targetHeaders(target: ProducerEventTarget): Record { + return target.kind === "node" + ? { "target-kind": "node", "target-id": String(target.value) } + : { "target-kind": target.kind }; +} + +let currentProducer: number | undefined; + +async function run(): Promise { + const root = document.getElementById("app"); + if (!root) throw new Error("#app not found"); + + const events = new Channel(); + const producer = await invoke("spawn_producer", { + name: "dioxus-todomvc", + events, + }); + currentProducer = producer; + + const driver = createDriver({ + root, + // `externalLinks: true` because the TodoMVC footer links out to real + // http(s) URLs; safe here specifically because `main.rs`'s + // `on_navigation` refuses to follow any of them — the producer only + // ever *names* a link, the host decides whether to honor it. + policy: desktopPolicy({ relativeHref: true, externalLinks: true }), + defaultPreventDefault: true, + onError: showError, + handleEvent(target, nameRef, payload) { + return invoke("send_event", payload, { + headers: { + producer: String(producer), + name: String(nameRef), + ...targetHeaders(target), + }, + }); + }, + }); + + events.onmessage = (msg) => { + if (msg.type === "closed") { + if (msg.error) showError(new Error(msg.error)); + return; + } + const target = msg.target; + let result: unknown = null; + switch (msg.kind) { + case "client-rect": + result = driver.queries.getClientRect(target) ?? null; + break; + case "scroll-offset": + result = driver.queries.getScrollOffset(target) ?? null; + break; + case "scroll-size": + result = driver.queries.getScrollSize(target) ?? null; + break; + case "focus": + result = driver.queries.setFocus(target, msg.focus ?? false); + break; + } + invoke("answer_query", { producer, id: msg.id, result }).catch(showError); + }; + + const latency: DesktopStreamDom["latency"] = { + reads: 0, + lastReadMs: 0, + maxPushMs: 0, + totalPushMs: 0, + }; + let resolveReady!: () => void; + const ready = new Promise((resolve) => { + resolveReady = resolve; + }); + globalThis.__streamDomDesktop = { + mounted: false, + ready, + stats: driver.stats, + latency, + }; + + for (;;) { + let buf: ArrayBuffer; + const t0 = performance.now(); + try { + buf = await invoke("read_chunk", { producer }); + } catch (err) { + // The producer died (trap, limit, or `kill_producer` already + // called elsewhere) — nothing more to read. + showError(err); + break; + } + latency.reads++; + latency.lastReadMs = performance.now() - t0; + const pushStart = performance.now(); + try { + driver.push(new Uint8Array(buf)); + } catch (err) { + // `push` throws on a protocol/policy violation: stop feeding, + // dispose, and tell the host side to tear the producer down — a + // `read_chunk` after this would wait on an ack that will never + // come (see bridge.rs's `PendingChunk` doc). + showError(err); + driver.dispose(); + await invoke("kill_producer", { producer }).catch(() => {}); + break; + } + const pushMs = performance.now() - pushStart; + latency.totalPushMs += pushMs; + if (pushMs > latency.maxPushMs) latency.maxPushMs = pushMs; + if (!globalThis.__streamDomDesktop!.mounted) { + globalThis.__streamDomDesktop!.mounted = true; + resolveReady(); + } + } +} + +addEventListener("beforeunload", () => { + // Best-effort: nothing awaits this, the page is going away regardless. + if (currentProducer !== undefined) { + invoke("kill_producer", { producer: currentProducer }).catch(() => {}); + } +}); + +run().catch(showError); diff --git a/host/stream-dom-host/Cargo.toml b/host/stream-dom-host/Cargo.toml new file mode 100644 index 0000000..28e0e69 --- /dev/null +++ b/host/stream-dom-host/Cargo.toml @@ -0,0 +1,31 @@ +[package] +name = "stream-dom-host" +version = "0.1.0" +edition.workspace = true +license.workspace = true +publish.workspace = true + +[dependencies] +# `component-model-async` pulls in `async` + `component-model`; `cranelift` +# and `runtime` are the compile and execute halves. Everything else the +# default feature set carries (gc, cache, profiling, coredump, wat, ...) is +# unused here, hence `default-features = false`. +wasmtime = { version = "47", default-features = false, features = [ + "component-model-async", + # `Error::from_anyhow`: wasmtime 47 has its own `Error` type, and the + # public API of this crate speaks `anyhow`. + "anyhow", + "cranelift", + "runtime", + "std", +] } +# p2 WASI: the guest imports `wasi:{io,clocks,cli,random}@0.2.9`. +wasmtime-wasi = "47" +anyhow = "1" +tokio = { version = "1", features = ["rt", "sync", "time", "macros"] } +log = "0.4" + +[dev-dependencies] +stream-dom-proto = { path = "../../crates/stream-dom-proto" } +prost = { version = "0.14", default-features = false, features = ["derive"] } +tokio = { version = "1", features = ["rt-multi-thread", "sync", "time", "macros"] } diff --git a/host/stream-dom-host/src/bindings.rs b/host/stream-dom-host/src/bindings.rs new file mode 100644 index 0000000..b62cab2 --- /dev/null +++ b/host/stream-dom-host/src/bindings.rs @@ -0,0 +1,38 @@ +//! `bindgen!` output for the `polymorph:stream-dom` `producer` world. +//! +//! The world's imports (`queries`, the `dom-event` resource) become host +//! traits implemented in [`crate::state`]; its exports (`run`, +//! `handle-event`) become typed concurrent calls used by +//! [`crate::producer`]. +//! +//! Both directions are bound `async`: every function in the world is an +//! `async func` in WIT except the two `dom-event` methods, and the +//! component-model async ABI is what `run`'s `stream` return and +//! `handle-event`'s concurrency with the live stream require. + +#[allow(missing_docs, reason = "generated code")] +mod generated { + wasmtime::component::bindgen!({ + path: "../../wit", + world: "producer", + imports: { + // `async` for the component-model async ABI, `store` for + // `Accessor` access to the store data (the bridge and the + // `ResourceTable`), `trappable` so a host error surfaces as a + // trap and kills the producer rather than being swallowed. + default: async | store | trappable, + // `prevent-default` / `stop-propagation` are plain `func`s and + // are no-ops here (see `crate::DomEvent`); binding them + // synchronously keeps them callable from the handler's + // synchronous prefix without an ABI round trip. + "polymorph:stream-dom/events@0.1.0.[method]dom-event.prevent-default": trappable, + "polymorph:stream-dom/events@0.1.0.[method]dom-event.stop-propagation": trappable, + }, + exports: { default: async }, + with: { + "polymorph:stream-dom/events@0.1.0.dom-event": crate::DomEvent, + }, + }); +} + +pub use self::generated::{polymorph::stream_dom, Producer as ProducerBindings}; diff --git a/host/stream-dom-host/src/consumer.rs b/host/stream-dom-host/src/consumer.rs new file mode 100644 index 0000000..9e0c9ff --- /dev/null +++ b/host/stream-dom-host/src/consumer.rs @@ -0,0 +1,114 @@ +//! The read end of a producer's mutation stream, gated on the receiver's ack. + +use std::pin::Pin; +use std::sync::Arc; +use std::task::{Context, Poll}; + +use wasmtime::component::{Source, StreamConsumer, StreamResult}; +use wasmtime::StoreContextMut; + +use crate::{Ctx, HostBridge}; + +/// Forwards each guest write to [`HostBridge::apply`] and withholds the +/// write's completion until the receiver has applied it. +/// +/// The mechanism is the one `StreamConsumer::poll_consume` documents under +/// "Backpressure": take the items, then return `Poll::Pending`, which tells +/// wasmtime to delay the `COMPLETED` event to the writer. The guest's +/// `stream.write` therefore does not resolve until `apply` does — the strict +/// rendezvous [`HostBridge::apply`] promises. +pub(crate) struct AckConsumer { + bridge: Arc, + /// The `apply` call for the bytes already taken out of `source`. Held + /// across polls: once taken, bytes cannot be put back, so the only + /// correct thing to do is finish delivering them. + inflight: Option>>, +} + +impl AckConsumer { + pub(crate) fn new(bridge: Arc) -> Self { + Self { + bridge, + inflight: None, + } + } +} + +impl StreamConsumer for AckConsumer { + type Item = u8; + + fn poll_consume( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + store: StoreContextMut, + source: Source<'_, u8>, + finish: bool, + ) -> Poll> { + let this = self.get_mut(); // safe: AckConsumer is Unpin + + // An apply from a previous poll is still running. Finish it before + // looking at `source` — the bytes it carries are already taken, and + // `finish` does not change that: with no out-of-band channel to + // report a partial application, interrupting delivery would leave + // the receiver's DOM in a state no one can describe. `poll_consume`'s + // docs name this the usually-preferable choice. + if let Some(fut) = this.inflight.as_mut() { + return match fut.as_mut().poll(cx) { + Poll::Pending => Poll::Pending, + Poll::Ready(result) => { + this.inflight = None; + // An apply error is unrecoverable at this level: there + // is no `future` in the WIT to report it on, so trap. + Poll::Ready(trap(result)) + } + }; + } + + let mut direct = source.as_direct(store); + // One `poll_consume` call is one guest write, and `remaining` is + // that whole write's buffer: taking all of it keeps the guest's + // one-write-per-batch framing intact end to end. Splitting would + // hand the receiver a partial batch; merging is impossible here + // anyway, since a second write cannot start until this one + // completes. + let chunk = direct.remaining().to_vec(); + if chunk.is_empty() { + return if finish { + // Nothing taken and the writer is cancelling: the only + // result allowed without taking an item. + Poll::Ready(Ok(StreamResult::Cancelled)) + } else { + // A zero-length write is a legal readiness probe. Reporting + // it consumed is correct because the next call can always + // accept an item; returning `Pending` would park forever, + // as nothing external would ever wake this task. + Poll::Ready(Ok(StreamResult::Completed)) + }; + } + direct.mark_read(chunk.len()); + + let mut fut = this.bridge_apply(chunk); + match fut.as_mut().poll(cx) { + Poll::Pending => { + this.inflight = Some(fut); + Poll::Pending + } + Poll::Ready(result) => Poll::Ready(trap(result)), + } + } +} + +/// A failed `apply` becomes a wasmtime trap: the guest's DOM and the +/// receiver's have diverged, and no WIT signature can carry the news back. +fn trap(result: anyhow::Result<()>) -> wasmtime::Result { + result + .map(|()| StreamResult::Completed) + .map_err(wasmtime::Error::from_anyhow) +} + +impl AckConsumer { + fn bridge_apply(&self, chunk: Vec) -> crate::BoxFuture<'static, anyhow::Result<()>> { + let bridge = self.bridge.clone(); + Box::pin(async move { bridge.apply(chunk).await }) + } +} diff --git a/host/stream-dom-host/src/lib.rs b/host/stream-dom-host/src/lib.rs new file mode 100644 index 0000000..b709c6e --- /dev/null +++ b/host/stream-dom-host/src/lib.rs @@ -0,0 +1,405 @@ +//! Wasmtime host for `polymorph:stream-dom` producers. +//! +//! One [`Host`] per process owns the engine and a linker that satisfies the +//! `producer` world (`queries`, the `dom-event` resource) plus WASI p2. Each +//! [`Producer`] is one component instance in its own [`Store`], running on +//! its own tokio task, whose mutation stream is forwarded chunk by chunk to +//! a [`HostBridge`] — the receiver as the host sees it, implemented by an +//! embedder (a Tauri layer talking to a webview) and by tests. +//! +//! See `docs/design.md` for the protocol; this crate is the "in-process +//! (component)" column of its Transports table, with the receiver one hop +//! further out than the table assumes. + +mod bindings; +mod consumer; +mod producer; + +use std::sync::Arc; +use std::time::Duration; + +use anyhow::Result; +use wasmtime::component::{Accessor, HasData, Linker, ResourceTable}; +use wasmtime::{Config, Engine, StoreLimits, StoreLimitsBuilder}; +use wasmtime_wasi::{WasiCtx, WasiCtxBuilder, WasiCtxView, WasiView}; + +pub use producer::{EventTarget, Producer}; +pub use wasmtime::component::Component; + +/// A DOM rect, flattened from the WIT's `rect { origin, size }`. +#[derive(Clone, Copy, Debug, PartialEq)] +pub struct Rect { + pub x: f64, + pub y: f64, + pub width: f64, + pub height: f64, +} + +#[derive(Clone, Copy, Debug, PartialEq)] +pub struct Point { + pub x: f64, + pub y: f64, +} + +#[derive(Clone, Copy, Debug, PartialEq)] +pub struct Size { + pub width: f64, + pub height: f64, +} + +/// The receiver side, as the host sees it. +/// +/// Every method is a host import the guest may sync-lower and block on, so +/// an implementation is free to take as long as a round trip to a webview +/// costs — see `docs/design.md` "Reads are `async`-typed host imports". +/// +/// Boxed futures rather than `async fn` in trait: this is used as +/// `Arc`, and async-fn-in-trait is not object safe. +pub trait HostBridge: Send + Sync + 'static { + /// Deliver one chunk of stream bytes — one guest write, normally one + /// batch — and return once the receiver has **applied** it. + /// + /// This is a strict rendezvous: the guest's `stream.write` does not + /// complete until this future resolves, so the guest can never run + /// ahead of the DOM. `docs/design.md` "Batches are framed by a `commit` + /// flag" makes this an invariant rather than a nicety — queries observe + /// every committed batch, which is only true if application is ordered + /// before the write that follows it. + /// + /// An `Err` traps the guest: the producer dies and [`Producer::closed`] + /// resolves with the cause. + fn apply(&self, chunk: Vec) -> BoxFuture<'_, Result<()>>; + + fn get_client_rect(&self, target: u32) -> BoxFuture<'_, Option>; + fn get_scroll_offset(&self, target: u32) -> BoxFuture<'_, Option>; + fn get_scroll_size(&self, target: u32) -> BoxFuture<'_, Option>; + fn set_focus(&self, target: u32, focus: bool) -> BoxFuture<'_, bool>; +} + +/// The future type [`HostBridge`] methods return. `Send` because a +/// producer's store runs on a tokio task. +pub type BoxFuture<'a, T> = std::pin::Pin + Send + 'a>>; + +/// Per-producer resource bounds. +#[derive(Clone, Copy, Debug)] +pub struct Limits { + /// Cap on a producer's linear memory. Exceeding it traps the guest, + /// which surfaces through [`Producer::closed`]. + pub memory_bytes: usize, + /// How often the engine's epoch advances. Each tick makes a running + /// guest yield to the tokio executor once, so one producer spinning in + /// a render loop cannot starve the others sharing the runtime. + pub epoch_tick: Duration, +} + +impl Default for Limits { + fn default() -> Self { + Self { + memory_bytes: 64 * 1024 * 1024, + epoch_tick: Duration::from_millis(10), + } + } +} + +/// The host-defined `dom-event` resource. +/// +/// Deliberately stateless: `prevent-default` and `stop-propagation` are +/// no-ops here. This tier's receiver is a webview one hop away, so the +/// guest's synchronous prefix cannot land an imperative verdict inside the +/// browser's listener — only the declarative flags on `add-listener` can +/// cross that boundary (`docs/design.md` "Events", option C). Accepting the +/// calls and ignoring them is what keeps a producer written against option +/// A running unmodified. +pub struct DomEvent; + +/// Engine, linker and limits shared by every producer in the process. +/// +/// Compilation is expensive and per-process; instantiation is per-producer. +pub struct Host { + engine: Engine, + linker: Arc>, + limits: Limits, + /// Aborts the epoch ticker when the host goes away. + ticker: tokio::task::JoinHandle<()>, +} + +impl Drop for Host { + fn drop(&mut self) { + self.ticker.abort(); + } +} + +impl Host { + /// Build the engine and linker. + /// + /// Must be called from within a tokio runtime: the epoch ticker is a + /// spawned task, tied to this host's lifetime. + pub fn new(limits: Limits) -> Result { + let mut config = Config::new(); + // The producer world is async through and through: `run` returns a + // `stream` and `handle-event` runs concurrently with it. + // (`Config::async_support` is deprecated and a no-op in wasmtime 47: + // async is implied by the `async` cargo feature.) + config.wasm_component_model_async(true); + // Required by `run_concurrent`, `call_concurrent` and `StreamReader`. + config.concurrency_support(true); + // Preemption for the cooperative scheduler: see `Limits::epoch_tick`. + config.epoch_interruption(true); + + let engine = Engine::new(&config)?; + + let mut linker = Linker::new(&engine); + wasmtime_wasi::p2::add_to_linker_async(&mut linker)?; + // Defines `queries` whether or not a given component imports it: a + // component that never calls a read has the import stripped, and a + // linker definition nothing imports is simply unused. + bindings::ProducerBindings::add_to_linker::(&mut linker, |ctx| ctx)?; + + let ticker = { + let engine = engine.clone(); + let tick = limits.epoch_tick; + tokio::spawn(async move { + let mut interval = tokio::time::interval(tick); + loop { + interval.tick().await; + engine.increment_epoch(); + } + }) + }; + + Ok(Self { + engine, + linker: Arc::new(linker), + limits, + ticker, + }) + } + + /// Compile a component. The result is `Send + Sync` and may be + /// instantiated into any number of producers. + pub fn compile(&self, wasm: &[u8]) -> Result { + Ok(Component::new(&self.engine, wasm)?) + } +} + +/// Per-store state: WASI, the resource table the `dom-event` resources live +/// in, the bridge every host import forwards to, and the memory limiter. +pub(crate) struct Ctx { + wasi: WasiCtx, + table: ResourceTable, + bridge: Arc, + limits: StoreLimits, +} + +impl Ctx { + fn new(bridge: Arc, limits: &Limits) -> Self { + Self { + // No preopens, no network, no env, no args: a producer needs a + // clock and randomness, nothing else. + wasi: WasiCtxBuilder::new() + .stdout(LogSink::new("stdout")) + .stderr(LogSink::new("stderr")) + .build(), + table: ResourceTable::new(), + bridge, + limits: StoreLimitsBuilder::new() + .memory_size(limits.memory_bytes) + // A component is many core module instances (the adapter, + // the guest, wit-bindgen's shims); these are headroom + // against a pathological one, not a tuned figure. + .instances(64) + .tables(64) + .build(), + } + } + + /// The store's memory limiter. `Store::limiter` wants a trait object. + pub(crate) fn limiter(&mut self) -> &mut dyn wasmtime::ResourceLimiter { + &mut self.limits + } +} + +impl WasiView for Ctx { + fn ctx(&mut self) -> WasiCtxView<'_> { + WasiCtxView { + ctx: &mut self.wasi, + table: &mut self.table, + } + } +} + +/// `HasData` marker tying the generated host traits to [`Ctx`]. +pub(crate) struct HasCtx; + +impl HasData for HasCtx { + type Data<'a> = &'a mut Ctx; +} + +/// Pull the bridge out of the store so the await happens without holding +/// store access — a query may block for a webview round trip, and the store +/// must stay available to the rest of the instance meanwhile. +fn bridge(accessor: &Accessor) -> Arc { + accessor.with(|mut access| access.get().bridge.clone()) +} + +impl bindings::stream_dom::types::Host for Ctx {} + +impl bindings::stream_dom::queries::HostWithStore for HasCtx { + async fn get_client_rect( + accessor: &Accessor, + target: u32, + ) -> wasmtime::Result> { + Ok(bridge(accessor).get_client_rect(target).await.map(|r| { + bindings::stream_dom::queries::Rect { + origin: bindings::stream_dom::queries::Point { x: r.x, y: r.y }, + size: bindings::stream_dom::queries::Size { + width: r.width, + height: r.height, + }, + } + })) + } + + async fn get_scroll_offset( + accessor: &Accessor, + target: u32, + ) -> wasmtime::Result> { + Ok(bridge(accessor) + .get_scroll_offset(target) + .await + .map(|p| bindings::stream_dom::queries::Point { x: p.x, y: p.y })) + } + + async fn get_scroll_size( + accessor: &Accessor, + target: u32, + ) -> wasmtime::Result> { + Ok(bridge(accessor).get_scroll_size(target).await.map(|s| { + bindings::stream_dom::queries::Size { + width: s.width, + height: s.height, + } + })) + } + + async fn set_focus( + accessor: &Accessor, + target: u32, + focus: bool, + ) -> wasmtime::Result { + Ok(bridge(accessor).set_focus(target, focus).await) + } +} + +impl bindings::stream_dom::queries::Host for Ctx {} + +impl bindings::stream_dom::events::Host for Ctx {} + +impl bindings::stream_dom::events::HostDomEvent for Ctx { + /// No-op; see [`DomEvent`]. + fn prevent_default( + &mut self, + _self_: wasmtime::component::Resource, + ) -> wasmtime::Result<()> { + Ok(()) + } + + /// No-op; see [`DomEvent`]. + fn stop_propagation( + &mut self, + _self_: wasmtime::component::Resource, + ) -> wasmtime::Result<()> { + Ok(()) + } +} + +impl bindings::stream_dom::events::HostDomEventWithStore for HasCtx { + async fn drop( + accessor: &Accessor, + rep: wasmtime::component::Resource, + ) -> wasmtime::Result<()> { + accessor.with(|mut access| access.get().table.delete(rep))?; + Ok(()) + } +} + +/// A WASI stdio sink that forwards whole lines to `log`. +/// +/// A producer's stdout is a diagnostic channel, not a data one; capturing it +/// keeps a panicking guest's message visible without giving the component a +/// real file descriptor. +struct LogSink { + which: &'static str, +} + +impl LogSink { + fn new(which: &'static str) -> Self { + Self { which } + } +} + +impl wasmtime_wasi::cli::IsTerminal for LogSink { + fn is_terminal(&self) -> bool { + false + } +} + +impl wasmtime_wasi::cli::StdoutStream for LogSink { + fn async_stream(&self) -> Box { + Box::new(LogWriter { + which: self.which, + line: Vec::new(), + }) + } +} + +/// Buffers until a newline so one guest `write` of a partial line does not +/// become one log record. +struct LogWriter { + which: &'static str, + line: Vec, +} + +impl LogWriter { + fn flush_lines(&mut self) { + while let Some(nl) = self.line.iter().position(|&b| b == b'\n') { + let rest = self.line.split_off(nl + 1); + let line = std::mem::replace(&mut self.line, rest); + let line = String::from_utf8_lossy(&line[..line.len() - 1]); + log::warn!("producer {}: {line}", self.which); + } + } +} + +impl tokio::io::AsyncWrite for LogWriter { + fn poll_write( + self: std::pin::Pin<&mut Self>, + _cx: &mut std::task::Context<'_>, + buf: &[u8], + ) -> std::task::Poll> { + let this = self.get_mut(); + this.line.extend_from_slice(buf); + this.flush_lines(); + std::task::Poll::Ready(Ok(buf.len())) + } + + fn poll_flush( + self: std::pin::Pin<&mut Self>, + _cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + std::task::Poll::Ready(Ok(())) + } + + fn poll_shutdown( + self: std::pin::Pin<&mut Self>, + _cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + let this = self.get_mut(); + if !this.line.is_empty() { + let line = String::from_utf8_lossy(&this.line); + log::warn!("producer {}: {line}", this.which); + this.line.clear(); + } + std::task::Poll::Ready(Ok(())) + } +} diff --git a/host/stream-dom-host/src/producer.rs b/host/stream-dom-host/src/producer.rs new file mode 100644 index 0000000..ed17bf0 --- /dev/null +++ b/host/stream-dom-host/src/producer.rs @@ -0,0 +1,254 @@ +//! One component instance: a store, a tokio task owning it, and the handle +//! the embedder drives it through. + +use std::sync::Arc; + +use anyhow::{anyhow, Result}; +use tokio::sync::{mpsc, oneshot, watch}; +use wasmtime::component::{Component, Resource}; +use wasmtime::Store; + +use crate::bindings::{stream_dom, ProducerBindings}; +use crate::consumer::AckConsumer; +use crate::{Ctx, DomEvent, Host, HostBridge}; + +/// What a listener was registered on. Mirrors the WIT `event-target`. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum EventTarget { + Node(u32), + Window, + Document, +} + +impl From for stream_dom::types::EventTarget { + fn from(t: EventTarget) -> Self { + match t { + EventTarget::Node(id) => Self::Node(id), + EventTarget::Window => Self::Window, + EventTarget::Document => Self::Document, + } + } +} + +/// A command to the actor. `Shutdown` is explicit rather than "drop the +/// sender" because [`Producer::shutdown`] must be callable through a shared +/// handle and must be able to wait for the teardown it asked for. +enum Cmd { + Dispatch(Dispatch), + Shutdown, +} + +/// One event dispatch, with the channel its completion is reported on. +struct Dispatch { + target: EventTarget, + name: u32, + payload: Vec, + done: oneshot::Sender>>, +} + +/// A live producer. +/// +/// The instance itself lives on a tokio task — a `Store` is `Send` but not +/// `Sync`, and the store must stay available to the event loop between +/// calls, so it is owned by an actor rather than shared behind a lock. +pub struct Producer { + events: mpsc::UnboundedSender, + /// `None` until the actor finishes; then the outcome, once. + death: watch::Receiver>>>, +} + +impl Producer { + /// Instantiate `component` and call `run(false)`, piping the returned + /// stream to `bridge`. + /// + /// Returns once `run` has returned its stream — that is, as soon as the + /// channel exists, which is strictly before the first + /// [`HostBridge::apply`] resolves and usually before the first one + /// starts. A caller may therefore dispatch events immediately; they + /// queue behind the instance lock like any other export call. + pub async fn spawn( + host: &Host, + component: &Component, + bridge: Arc, + ) -> Result { + let (events, event_rx) = mpsc::unbounded_channel(); + let (ready_tx, ready_rx) = oneshot::channel(); + let (death_tx, death) = watch::channel(None); + + let engine = host.engine.clone(); + let linker = host.linker.clone(); + let component = component.clone(); + let limits = host.limits; + + tokio::spawn(async move { + let mut store = Store::new(&engine, Ctx::new(bridge, &limits)); + store.limiter(Ctx::limiter); + // Yield to the executor on every epoch tick rather than trap: + // a producer that renders for a long time is doing its job, it + // just must not monopolize the runtime while doing it. + store.set_epoch_deadline(1); + store.epoch_deadline_async_yield_and_update(1); + + let outcome = run_instance(&mut store, &linker, &component, event_rx, ready_tx).await; + let _ = death_tx.send(Some(outcome.map_err(Arc::new))); + }); + + match ready_rx.await { + Ok(()) => Ok(Self { events, death }), + // The actor died before `run` returned; report why rather than + // "the channel closed". + Err(_) => { + let mut death = death; + let cause = wait_death(&mut death).await; + Err(match cause { + Err(e) => anyhow!("producer failed to start: {e:#}"), + Ok(()) => anyhow!("producer stopped before `run` returned"), + }) + } + } + } + + /// Dispatch one event and resolve when the guest's handler has + /// returned. + /// + /// Dispatches are serialized: the actor awaits each `handle-event` + /// before taking the next. The instance's exclusive lock would serialize + /// them at the backpressure gate anyway, and doing it here keeps them in + /// the order the receiver observed them. + pub async fn handle_event( + &self, + target: EventTarget, + name: u32, + payload: Vec, + ) -> Result<()> { + let (done, wait) = oneshot::channel(); + self.events + .send(Cmd::Dispatch(Dispatch { + target, + name, + payload, + done, + })) + .map_err(|_| anyhow!("producer is dead"))?; + match wait.await { + Ok(Ok(())) => Ok(()), + Ok(Err(e)) => Err(anyhow!("{e:#}")), + Err(_) => Err(anyhow!("producer died during dispatch")), + } + } + + /// Tear the producer down, returning once it has stopped. Idempotent: + /// on an already-dead producer the send fails and the recorded outcome + /// is already there. + /// + /// Ending the actor's loop returns from the store's concurrent scope and + /// drops the store, which is the only way wasmtime 47 offers to cancel + /// guest tasks still in flight (`Func::call_concurrent`, "Cancellation"). + pub async fn shutdown(&self) { + let _ = self.events.send(Cmd::Shutdown); + let mut death = self.death.clone(); + let _ = wait_death(&mut death).await; + } + + /// Resolve when the producer stops: `Ok` for an orderly shutdown, `Err` + /// with the cause for a trap, a failed `apply`, or an exhausted memory + /// limit. The embedder reports the cause to its receiver. + pub async fn closed(&self) -> Result<(), Arc> { + let mut death = self.death.clone(); + wait_death(&mut death).await + } +} + +async fn wait_death( + death: &mut watch::Receiver>>>, +) -> Result<(), Arc> { + loop { + if let Some(outcome) = death.borrow_and_update().clone() { + return outcome; + } + if death.changed().await.is_err() { + // The actor task was cancelled without recording an outcome. + return Err(Arc::new(anyhow!("producer task ended unexpectedly"))); + } + } +} + +/// The actor body: instantiate, start the stream, then serve dispatches +/// until the handle goes away. +async fn run_instance( + store: &mut Store, + linker: &wasmtime::component::Linker, + component: &Component, + mut events: mpsc::UnboundedReceiver, + ready: oneshot::Sender<()>, +) -> Result<()> { + // Instantiation is outside the concurrent scope because it needs the + // store itself, and `Accessor` only lends it a synchronous view. + let bindings = ProducerBindings::instantiate_async(&mut *store, component, linker).await?; + + store + .run_concurrent(async move |accessor| -> Result<()> { + // `hydrate: false` — this tier mounts into an empty root; a + // prerendered one would be the embedder's choice to expose. + let stream = bindings.call_run(accessor, false).await?; + let bridge = accessor.with(|mut access| access.get().bridge.clone()); + accessor.with(|access| stream.pipe(access, AckConsumer::new(bridge)))?; + + // From here the stream is live: the event loop polls the + // consumer while this task waits on dispatches. + let _ = ready.send(()); + + // Ends on `Shutdown` or when the last handle drops: either way + // the scope returns and the store is dropped by the caller. + while let Some(cmd) = events.recv().await { + let dispatch = match cmd { + Cmd::Dispatch(d) => d, + Cmd::Shutdown => break, + }; + let result = dispatch_event(accessor, &bindings, &dispatch).await; + match result { + Ok(()) => { + let _ = dispatch.done.send(Ok(())); + } + Err(e) => { + // A trapped handler kills the instance; the caller + // waiting on this dispatch and `closed()` both get + // the same cause. + let e = Arc::new(e); + let _ = dispatch.done.send(Err(e.clone())); + return Err(anyhow!("{e:#}")); + } + } + } + Ok(()) + }) + .await? +} + +async fn dispatch_event( + accessor: &wasmtime::component::Accessor, + bindings: &ProducerBindings, + dispatch: &Dispatch, +) -> Result<()> { + // The resource exists for exactly one dispatch: the WIT lends it to the + // handler's synchronous prefix, and nothing may name it afterwards. + let owned = accessor.with(|mut access| access.get().table.push(DomEvent))?; + let borrow = Resource::new_borrow(owned.rep()); + + let result = bindings + .call_handle_event( + accessor, + dispatch.target.into(), + dispatch.name, + dispatch.payload.clone(), + borrow, + ) + .await; + + // Deleted whether or not the call succeeded: a trap leaves the table + // entry behind otherwise, and the store may outlive the failed call. + let deleted = accessor.with(|mut access| access.get().table.delete(owned)); + result?; + deleted?; + Ok(()) +} diff --git a/host/stream-dom-host/tests/dioxus_todomvc.rs b/host/stream-dom-host/tests/dioxus_todomvc.rs new file mode 100644 index 0000000..2f4a10e --- /dev/null +++ b/host/stream-dom-host/tests/dioxus_todomvc.rs @@ -0,0 +1,487 @@ +//! End-to-end: the real TodoMVC producer component, hosted on wasmtime, +//! driven through a fake receiver. +//! +//! What this pins down is the semantics the crate exists for — the ack +//! rendezvous, event dispatch reaching a real handler, teardown, and limit +//! failures surfacing rather than hanging — against a component nobody +//! wrote for this host. + +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; + +use anyhow::Result; +use prost::Message as _; +use stream_dom_host::{ + BoxFuture, EventTarget, Host, HostBridge, Limits, Point, Producer, Rect, Size, +}; +use stream_dom_proto as proto; + +const COMPONENT: &str = "build/dioxus-todomvc.component.wasm"; + +fn component_bytes() -> Vec { + // CARGO_MANIFEST_DIR is host/stream-dom-host; the component lives in the + // repo's build/ directory. + let path = std::path::Path::new(env!("CARGO_MANIFEST_DIR")) + .join("../..") + .join(COMPONENT); + std::fs::read(&path).unwrap_or_else(|e| { + panic!("{COMPONENT} is missing ({e}); build it with `just host-component`") + }) +} + +/// One recorded `apply`: the bytes, and when the call started, so the ack +/// gate can be observed in time rather than only in order. +/// +/// Recorded when the call *starts*, not when it finishes: a test that +/// stalls an apply still needs to read the chunk it stalled on. +struct Applied { + chunk: Vec, + started: Instant, +} + +/// A receiver that records everything and can be told to stall. +#[derive(Default)] +struct Recorder { + applied: Vec, + /// How long the *next* `apply` (by index) should hold before returning. + stall_first: Option, + /// Number of `apply` calls that have started but not finished. + in_flight: usize, + /// Set if two applies were ever in flight at once — the property the + /// rendezvous forbids. + overlapped: bool, +} + +#[derive(Default)] +struct FakeBridge { + state: Mutex, + notify: tokio::sync::Notify, +} + +impl FakeBridge { + fn stalling(first: Duration) -> Arc { + Arc::new(Self { + state: Mutex::new(Recorder { + stall_first: Some(first), + ..Recorder::default() + }), + notify: tokio::sync::Notify::new(), + }) + } + + fn chunks(&self) -> Vec> { + self.state + .lock() + .unwrap() + .applied + .iter() + .map(|a| a.chunk.clone()) + .collect() + } + + fn applied_count(&self) -> usize { + self.state.lock().unwrap().applied.len() + } + + /// Every frame delivered so far, in order, across all chunks. + fn frames(&self) -> Vec { + self.chunks() + .iter() + .flat_map(|c| decode_frames(c)) + .collect() + } + + /// Wait until at least `n` applies have *started*, or time out. + async fn wait_for_applies(&self, n: usize, timeout: Duration) -> bool { + tokio::time::timeout(timeout, async { + loop { + if self.applied_count() >= n { + return; + } + self.notify.notified().await; + } + }) + .await + .is_ok() + } +} + +impl HostBridge for FakeBridge { + fn apply(&self, chunk: Vec) -> BoxFuture<'_, Result<()>> { + Box::pin(async move { + let stall = { + let mut state = self.state.lock().unwrap(); + if state.in_flight > 0 { + state.overlapped = true; + } + state.in_flight += 1; + state.applied.push(Applied { + chunk, + started: Instant::now(), + }); + state.stall_first.take() + }; + self.notify.notify_waiters(); + if let Some(stall) = stall { + tokio::time::sleep(stall).await; + } + self.state.lock().unwrap().in_flight -= 1; + Ok(()) + }) + } + + fn get_client_rect(&self, _target: u32) -> BoxFuture<'_, Option> { + Box::pin(async { None }) + } + fn get_scroll_offset(&self, _target: u32) -> BoxFuture<'_, Option> { + Box::pin(async { None }) + } + fn get_scroll_size(&self, _target: u32) -> BoxFuture<'_, Option> { + Box::pin(async { None }) + } + fn set_focus(&self, _target: u32, _focus: bool) -> BoxFuture<'_, bool> { + Box::pin(async { false }) + } +} + +/// Split a chunk into `Frame`s: varint byte length, then one `Frame`, +/// repeated (proto/stream-dom.proto, "Stream layout"). A frame may straddle +/// a chunk in general; this asserts it does not, which holds because the +/// guest writes whole batches. +fn decode_frames(mut buf: &[u8]) -> Vec { + let mut frames = Vec::new(); + while !buf.is_empty() { + let len = prost::encoding::decode_varint(&mut buf).expect("frame length varint") as usize; + assert!(len <= buf.len(), "frame straddles a chunk boundary"); + let (frame, rest) = buf.split_at(len); + frames.push(proto::Frame::decode(frame).expect("decode Frame")); + buf = rest; + } + frames +} + +/// The interned string table built from the `Intern` frames seen so far. +fn interns(frames: &[proto::Frame]) -> std::collections::HashMap { + frames + .iter() + .filter_map(|f| match &f.op { + Some(proto::frame::Op::Intern(i)) => Some((i.id, i.s.clone())), + _ => None, + }) + .collect() +} + +fn listeners(frames: &[proto::Frame]) -> Vec { + frames + .iter() + .filter_map(|f| match &f.op { + Some(proto::frame::Op::AddListener(a)) => a.listener, + _ => None, + }) + .collect() +} + +async fn host() -> Host { + Host::new(Limits::default()).expect("host") +} + +/// The new-todo input, as `(node, input str-ref, keydown str-ref)`. +/// +/// Identified as the node carrying both an `input` and a `keydown` +/// listener — TodoMVC's `TodoHeader` is the only place that pairs them — +/// rather than by class, which would depend on which template hole it +/// landed in. +fn new_todo_input(frames: &[proto::Frame]) -> (u32, u32, u32) { + let names = interns(frames); + let name_of = |slot: u32| names.get(&slot).cloned().unwrap_or_default(); + let ls = listeners(frames); + let node_of = |l: &proto::Listener| match l.target { + Some(proto::listener::Target::Id(id)) => Some(id), + _ => None, + }; + let keydown_nodes: Vec = ls + .iter() + .filter(|l| name_of(l.name) == "keydown") + .filter_map(node_of) + .collect(); + ls.iter() + .filter(|l| name_of(l.name) == "input") + .find_map(|l| { + let id = node_of(l)?; + keydown_nodes.contains(&id).then(|| { + let keydown = ls + .iter() + .find(|k| name_of(k.name) == "keydown" && node_of(k) == Some(id)) + .expect("keydown listener"); + (id, l.name, keydown.name) + }) + }) + .expect("no node carries both an `input` and a `keydown` listener") +} + +/// The `input` event the draft-text handler reads (`evt.value()`). +fn form_payload(value: &str) -> Vec { + proto::EventPayload { + family: Some(proto::event_payload::Family::Form(proto::FormData { + value: value.to_string(), + ..Default::default() + })), + } + .encode_to_vec() +} + +/// The `keydown` the handler tests with `evt.key() == Key::Enter`. +fn enter_payload() -> Vec { + proto::EventPayload { + family: Some(proto::event_payload::Family::Keyboard( + proto::KeyboardData { + key: "Enter".to_string(), + code: "Enter".to_string(), + ..Default::default() + }, + )), + } + .encode_to_vec() +} + +/// Type a draft and press Enter on the new-todo input. +async fn add_todo(producer: &Producer, frames: &[proto::Frame], text: &str) -> Result<()> { + let (node, input_name, keydown_name) = new_todo_input(frames); + producer + .handle_event(EventTarget::Node(node), input_name, form_payload(text)) + .await?; + producer + .handle_event(EventTarget::Node(node), keydown_name, enter_payload()) + .await?; + Ok(()) +} + +/// The first chunks are a well-formed mount: at least one batch ends in +/// `commit`, nodes are created (directly or by cloning a template), and the +/// producer registers listeners. +#[tokio::test(flavor = "multi_thread")] +async fn mount_produces_a_committed_batch_with_listeners() { + let host = host().await; + let component = host.compile(&component_bytes()).expect("compile"); + let bridge = Arc::new(FakeBridge::default()); + + let producer = Producer::spawn(&host, &component, bridge.clone()) + .await + .expect("spawn"); + assert!( + bridge.wait_for_applies(1, Duration::from_secs(10)).await, + "no batch arrived from the mount" + ); + + let frames = bridge.frames(); + assert!( + frames.iter().any(|f| f.commit), + "no batch was committed: {} frames", + frames.len() + ); + assert!( + frames.iter().any(|f| matches!( + f.op, + Some(proto::frame::Op::CreateElement(_)) | Some(proto::frame::Op::CloneTemplate(_)) + )), + "no create_element / clone_template op in the mount" + ); + assert!( + !listeners(&frames).is_empty(), + "the mount registered no listeners" + ); + + producer.shutdown().await; +} + +/// The rendezvous: a receiver that holds one `apply` holds the guest's +/// write, so a second `apply` cannot start until the first returns. +/// +/// TodoMVC writes its whole mount in one write and then parks (the WIT's +/// "a parked scheduler is a documented idle state"), so a second write has +/// to be provoked: an event is dispatched while the mount's apply is still +/// stalled. +/// +/// `spawn` itself must not be gated on the ack: it returns once `run` has +/// returned the stream, which is before the first `apply` resolves. +#[tokio::test(flavor = "multi_thread")] +async fn a_stalled_apply_blocks_the_next_one_but_not_spawn() { + let host = host().await; + let component = host.compile(&component_bytes()).expect("compile"); + let stall = Duration::from_millis(200); + let bridge = FakeBridge::stalling(stall); + + let before_spawn = Instant::now(); + let producer = Producer::spawn(&host, &component, bridge.clone()) + .await + .expect("spawn"); + let spawn_took = before_spawn.elapsed(); + // `run` returns its stream before any batch is applied, so spawn cannot + // have waited out the stall. + assert!( + spawn_took < stall, + "spawn took {spawn_took:?}, so it waited on the first apply" + ); + + assert!( + bridge.wait_for_applies(1, Duration::from_secs(10)).await, + "no batch arrived from the mount" + ); + // The mount's apply is stalling right now; this event's batch is the + // write that must queue behind it. + add_todo(&producer, &bridge.frames(), "write a host") + .await + .expect("dispatch"); + + assert!( + bridge.wait_for_applies(2, Duration::from_secs(10)).await, + "the event produced no second batch" + ); + producer.shutdown().await; + + let state = bridge.state.lock().unwrap(); + assert!( + !state.overlapped, + "two applies were in flight at once: the write was not gated on the ack" + ); + let gap = state.applied[1] + .started + .duration_since(state.applied[0].started); + assert!( + gap >= stall, + "the second apply started {gap:?} after the first, less than the {stall:?} stall" + ); +} + +/// Dispatch the two events TodoMVC's `TodoHeader` needs to add a row: an +/// `input` carrying the draft text (the handler reads `evt.value()`), then +/// a `keydown` with `key = "Enter"` — the same pair `web/e2e/todomvc_test.ts` +/// performs through a browser. A new todo row must then reach the receiver. +#[tokio::test(flavor = "multi_thread")] +async fn dispatching_input_then_enter_adds_a_todo() { + let host = host().await; + let component = host.compile(&component_bytes()).expect("compile"); + let bridge = Arc::new(FakeBridge::default()); + + let producer = Producer::spawn(&host, &component, bridge.clone()) + .await + .expect("spawn"); + assert!(bridge.wait_for_applies(1, Duration::from_secs(10)).await); + + let before = bridge.applied_count(); + add_todo(&producer, &bridge.frames(), "write a host") + .await + .expect("dispatch"); + + // The handler runs on the guest's scheduler task, so the resulting + // batch arrives after `handle-event` has already returned. + assert!( + bridge + .wait_for_applies(before + 1, Duration::from_secs(10)) + .await, + "no batch followed the Enter keydown" + ); + + let after: Vec = bridge.chunks()[before..] + .iter() + .flat_map(|c| decode_frames(c)) + .collect(); + assert!( + after.iter().any(|f| matches!( + f.op, + Some(proto::frame::Op::CreateElement(_)) + | Some(proto::frame::Op::CloneTemplate(_)) + | Some(proto::frame::Op::CreateText(_)) + )), + "the batch after Enter created no nodes" + ); + // The draft text itself must have crossed: either as a text node or as + // an interned string the row's ops refer to. + assert!( + interns(&bridge.frames()) + .values() + .any(|s| s == "write a host") + || after.iter().any(|f| matches!( + &f.op, + Some(proto::frame::Op::CreateText(t)) if t.text == "write a host" + )) + || after.iter().any(|f| matches!( + &f.op, + Some(proto::frame::Op::SetText(t)) if t.text == "write a host" + )), + "the todo's text never reached the receiver" + ); + + // The producer survived the round trip. + assert!( + tokio::time::timeout(Duration::from_millis(50), producer.closed()) + .await + .is_err(), + "the producer died during dispatch" + ); + + producer.shutdown().await; +} + +/// `shutdown` stops the producer and `closed` reports the orderly stop. +#[tokio::test(flavor = "multi_thread")] +async fn shutdown_resolves_closed() { + let host = host().await; + let component = host.compile(&component_bytes()).expect("compile"); + let bridge = Arc::new(FakeBridge::default()); + + let producer = Producer::spawn(&host, &component, bridge.clone()) + .await + .expect("spawn"); + assert!(bridge.wait_for_applies(1, Duration::from_secs(10)).await); + + producer.shutdown().await; + let closed = tokio::time::timeout(Duration::from_secs(5), producer.closed()) + .await + .expect("closed() did not resolve after shutdown"); + assert!(closed.is_ok(), "orderly shutdown reported {closed:?}"); + + // Idempotent. + producer.shutdown().await; +} + +/// A memory limit the guest cannot fit in must fail loudly — not panic, not +/// hang. Whether it fails at instantiation (`spawn` returns the error) or +/// once running (`closed` carries it) is not something the limit lets us +/// choose, so either is accepted. +#[tokio::test(flavor = "multi_thread")] +async fn an_impossible_memory_limit_fails_rather_than_hangs() { + let host = Host::new(Limits { + memory_bytes: 1024 * 1024, + ..Limits::default() + }) + .expect("host"); + let component = host.compile(&component_bytes()).expect("compile"); + let bridge = Arc::new(FakeBridge::default()); + + let spawned = tokio::time::timeout( + Duration::from_secs(30), + Producer::spawn(&host, &component, bridge.clone()), + ) + .await + .expect("spawn neither returned nor failed within 30s"); + + match spawned { + Err(e) => { + let msg = format!("{e:#}"); + assert!( + msg.contains("memory") || msg.contains("limit") || msg.contains("grow"), + "expected a memory-limit failure, got: {msg}" + ); + } + Ok(producer) => { + let closed = tokio::time::timeout(Duration::from_secs(30), producer.closed()) + .await + .expect("producer neither died nor reported within 30s"); + assert!( + closed.is_err(), + "a 1 MiB producer ran to an orderly stop; the limit did nothing" + ); + } + } +} diff --git a/justfile b/justfile index cf3b3a7..592ca86 100644 --- a/justfile +++ b/justfile @@ -8,14 +8,27 @@ default: check test check: cargo clippy --workspace --target wasm32-wasip2 -- -D warnings cargo clippy --manifest-path guests/web-sys/Cargo.toml --workspace --target wasm32-wasip2 -- -D warnings + cargo clippy --manifest-path host/Cargo.toml --workspace -- -D warnings deno task check -# Native unit tests (encoder, transcoder fixtures) + receiver tests. -test: +# Native unit tests (encoder, transcoder fixtures) + receiver tests + the +# wasmtime host, whose integration test runs the TodoMVC component. +test: host-component cargo test --workspace --exclude dioxus-todomvc cargo test --manifest-path guests/web-sys/Cargo.toml --workspace + cargo test --manifest-path host/Cargo.toml --workspace deno task test +# The one component `just test` needs. Same build as the `component` recipe +# without the translation step, which needs deno and the network. +host-component: + #!/usr/bin/env bash + set -euo pipefail + mkdir -p build + cargo build -p dioxus-todomvc --target wasm32-wasip2 --release + cp target/wasm32-wasip2/release/dioxus_todomvc.wasm build/dioxus-todomvc.component.wasm + wasm-tools validate --features component-model,cm-async build/dioxus-todomvc.component.wasm + # Build all demo + bench components into build/ and translate them at # build time (the demos ship no translator). components: (component "dioxus-todomvc" "Cargo.toml" "dioxus_todomvc") (component "dominator-todomvc" "guests/web-sys/Cargo.toml" "dominator_todomvc") (component "dioxus-bench" "Cargo.toml" "dioxus_bench") (component "dominator-bench" "guests/web-sys/Cargo.toml" "dominator_bench") @@ -54,3 +67,29 @@ bench-wire: site bench: site npx -y playwright@1.58 install chromium deno run -A bench/run.ts + +# Bundle the desktop app's frontend (host/desktop/dist/). +desktop-ui: + deno run -A host/desktop/build.ts + +# Build the Tauri desktop app: the TodoMVC component it embeds, its +# frontend bundle, then the Rust binary. `resource_dir()` in an unbundled +# dev/CI build resolves to the binary's own directory (tauri-utils +# `platform::resource_dir`'s "cargo output directory" case), so the +# component the binary loads via `BaseDirectory::Resource` has to be +# copied there by hand rather than relying on `tauri.conf.json`'s +# `bundle.resources` (that only fires for `cargo tauri build`, which this +# recipe deliberately does not invoke — the dispatch names a plain +# `cargo build --release`). +desktop: host-component desktop-ui + cargo build --manifest-path host/Cargo.toml -p stream-dom-desktop --release + mkdir -p host/target/release/components + cp build/dioxus-todomvc.component.wasm host/target/release/components/ + +# WebDriver smoke test against the built app (host/desktop/e2e/smoke.ts). +# No display on this machine: everything GUI runs under `xvfb-run -a`. +# Named explicitly, not `host/desktop/e2e/`: `deno test` only autodiscovers +# `*_test.ts`/`*.test.ts`, and `smoke.ts` (not `smoke_test.ts`) is the name +# this track's dispatch specified. +desktop-smoke: desktop + xvfb-run -a deno test -A host/desktop/e2e/smoke.ts diff --git a/receiver/src/driver.ts b/receiver/src/driver.ts index bb9b059..ad5f706 100644 --- a/receiver/src/driver.ts +++ b/receiver/src/driver.ts @@ -101,6 +101,25 @@ export interface DriverOptions { * rejections and strict-decode errors surface synchronously out of * `push` instead — see `Driver.push`. */ onError?(err: unknown): void; + /** Refuse real navigation out of the mount by default: every `submit` + * is `preventDefault()`-ed, and every `click` on an `` whose + * `href` is not a same-document fragment (`#...`) is too — installed + * as two root-level, capture-phase, non-passive listeners regardless + * of whether the producer registered anything at all (docs/design.md + * "Events", option C: "a per-event-type default policy at the + * receiver ... a registered `submit` listener implies preventDefault; + * a `click` listener on `` likewise"). This is deliberately + * BROADER than that phrasing: an unlistened `
`/`` is the + * dangerous case (real submit, real navigation, with no producer + * listener in the way to have its declarative flag inspected at all), + * so the receiver refuses by default rather than only when a listener + * happens to be registered. Off by default: the in-page polyengine + * mount uses option A (imperative `prevent-default`) and must keep + * today's behavior; a desktop/remote embedding with no imperative path + * sets this true so a `Dioxus`-style producer emitting no declarative + * flags at all (docs/design.md "Spike") does not submit every form and + * navigate every link for real. */ + defaultPreventDefault?: boolean; /** Deliver one event to the producer. `target`/`nameRef`/`payload` are * exactly what `handle-event` takes; `ev` is the live native Event, lent * for the synchronous prefix (the component glue wraps it in the WIT @@ -222,6 +241,50 @@ export function createDriver(opts: DriverOptions): Driver { } } + // -- default preventDefault (docs/design.md "Events", option C's + // per-event-type default; see `DriverOptions.defaultPreventDefault`'s + // doc for why this is two unconditional root-level listeners rather + // than a check inside `fire`) ------------------------------------------- + // + // Capture phase and explicitly non-passive: registered once, at + // construction, on `opts.root` itself, so neither the passive-listener + // "preventDefault is a no-op" hazard nor the delegated-listener refcount + // bookkeeping above ever comes into it — this fires (and can call + // `preventDefault`) before ANY producer-registered listener, delegated + // or direct, sees the event at all. + + function defaultSubmitHandler(e: Event): void { + e.preventDefault(); + } + + function defaultClickHandler(e: Event): void { + let node: Node | null = e.target as Node | null; + while (node) { + if (node.nodeType === 1) { + const el = node as Element; + if (el.tagName.toUpperCase() === "A" && el.hasAttribute("href")) { + if (!(el.getAttribute("href") ?? "").startsWith("#")) { + e.preventDefault(); + } + break; + } + } + if (node === opts.root) break; + node = node.parentNode; + } + } + + if (opts.defaultPreventDefault) { + opts.root.addEventListener("submit", defaultSubmitHandler, { + capture: true, + passive: false, + }); + opts.root.addEventListener("click", defaultClickHandler, { + capture: true, + passive: false, + }); + } + // -- event delegation ------------------------------------------------------- // // Bubbling listeners are delegated at `root`: one native listener per @@ -499,6 +562,14 @@ export function createDriver(opts: DriverOptions): Driver { if (disposed) return; disposed = true; gate.dispose(); + if (opts.defaultPreventDefault) { + opts.root.removeEventListener("submit", defaultSubmitHandler, { + capture: true, + }); + opts.root.removeEventListener("click", defaultClickHandler, { + capture: true, + }); + } for (const entry of rootListeners.values()) { opts.root.removeEventListener(entry.name, entry.handler, { capture: entry.capture, diff --git a/receiver/src/mod.ts b/receiver/src/mod.ts index a4a1d69..041e400 100644 --- a/receiver/src/mod.ts +++ b/receiver/src/mod.ts @@ -20,6 +20,8 @@ export { NativeDomReceiver } from "./native.ts"; export { encodePayload } from "./events.ts"; export { assertPolicyVersion, PolicyError, PolicySink } from "./policy.ts"; export type { Policy, PolicyOp } from "./policy.ts"; +export { desktopPolicy } from "./policy-desktop.ts"; +export type { DesktopPolicyOptions } from "./policy-desktop.ts"; export { DispatchGate } from "./dispatch.ts"; export { mount } from "./mount.ts"; export type { Mounted, MountOptions } from "./mount.ts"; diff --git a/receiver/src/policy-desktop.ts b/receiver/src/policy-desktop.ts new file mode 100644 index 0000000..53f8251 --- /dev/null +++ b/receiver/src/policy-desktop.ts @@ -0,0 +1,684 @@ +// The first concrete embedder policy (docs/design.md "Policy": "This +// protocol ships no allowlist ... What it ships is the seam"): a +// fail-closed vocabulary for an UNTRUSTED producer — a third-party plugin +// in a wasmtime sandbox — rendered into a privileged host webview. Any +// DOM-level escape (`