diff --git a/.mise/config.rust.toml b/.mise/config.rust.toml index adf1a7c6..00cf07b0 100644 --- a/.mise/config.rust.toml +++ b/.mise/config.rust.toml @@ -69,7 +69,7 @@ shell = "{{ vars.task_shell }}" description = "Build the no_std crates for a bare-metal target" run = """ features=et-ws-service/picoserve-hub,et-ws-server/picoserve-server -set -- -p et-org -p edge-toolkit -p et-ws-service -p et-ws-server +set -- -p et-org -p edge-toolkit -p et-ws-service -p et-ws-server -p et-ws-wasi-runner cargo build --lib --target thumbv8m.main-none-eabihf --no-default-features --features "$features" "$@" """ shell = "{{ vars.task_shell }}" diff --git a/CHANGELOG.md b/CHANGELOG.md index 18a2c37e..7d36225b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -42,6 +42,9 @@ Changelog 1.1.0. from the same `MODULES_*` variables, so every board's firmware embeds modules the same way. Its `board` writes what the board a target picks bakes in -- the Wi-Fi network, and the Pico 2 W's radio blobs and memory layout -- and returns the directives that link it. +- `et-ws-wasi-runner` builds as `no_std` + `alloc` without its new default `std` feature, and its `board` module + runs a guest on a microcontroller: compiled ahead of time with `engine::precompile` for the board's Pulley target, + and hosted through the firmware's `board::Platform` and `board::Transport`. - An opt-in `et-recovery-supervisor` service observes application progress and coordinates bounded Kubernetes recovery. - The Python SDK provides recovery lifecycle messages and integrity-checked checkpoint storage helpers. - A scenario can list `machines:` to span several hosts. The first is the server, running the hub and the collector, @@ -100,6 +103,11 @@ Changelog 1.1.0. - **Breaking:** `et-ws-server`'s actix server, its binary and its environment config are behind the new default `std` feature. A build with `default-features = false` and none of the features that imply `std` -- `modules`, `otlp`, `qr`, `storage`, `tls`, `websockify` -- now gets the `no_std` library and no binary; add `std` to keep the server. +- **Breaking:** `et-ws-wasi-runner`'s host runner, its binary and `run_module` are behind the new default `std` + feature, which `nn`, `webgpu` and `coverage` imply. A build with `default-features = false` and none of them now + gets the `no_std` board runner and no binary; add `std` to keep the host runner. `server_message_to_wit` moved from + `host::ws` to `convert`, and `Bucket` from `host::wasi_keyvalue` and `kv_not_implemented` from `host` to + `keyvalue`. - **Breaking:** `et-cli` refuses a scenario whose `module_paths:` entry, or a module directory found under one, has a name a shell would need quoted or that some filesystem could not hold, such as `my mod`, `mod$(id)` or `con`; rename the directory. Generated deployments no longer escape these paths. The check is the new `et-sanitise` crate, which diff --git a/Cargo.toml b/Cargo.toml index a57e1e05..ae76d7fd 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -306,7 +306,14 @@ wasm-bindgen-test = "0.3" # RUSTSEC-2026-0378 / RUSTSEC-2026-0379 only from 49.0.4, and RUSTSEC-2026-0420 and RUSTSEC-2026-0422 through # RUSTSEC-2026-0424 only from 49.0.5, on the 49 line. The major is held at 49 by wasi-webgpu-wasmtime, whose 0.4 # line requires wasmtime ^49. -wasmtime = { version = "49.0.5", features = ["async", "component-model"] } +# Without its default features, which need std, so a board's runner can depend on it; a host build turns them on. +wasmtime = { version = "49.0.5", default-features = false, features = [ + "async", + "component-model", + "component-model-async", + "pulley", + "runtime", +] } # The generator behind wasmtime's `bindgen!` macro, driven directly by et-int-gen. # Emitting the runner's host bindings as a checked-in source file keeps ws-wasi-runner self-contained: the # macro read WIT from `../../generated/specs/wit`, which `cargo package` cannot include, so the crate failed diff --git a/config/deny.toml b/config/deny.toml index 8bf62193..13d91199 100644 --- a/config/deny.toml +++ b/config/deny.toml @@ -199,6 +199,40 @@ ignore = [ # newest release. It runs at compile time inside that macro and reports its errors; none of it is in the firmware. # Drop this when pio-proc leaves it. { id = "RUSTSEC-2026-0173", reason = "expires 2026-11-13" }, + # The npm: AsyncAPI generator tools -- the entries below share one setting, so it is described once, here. + # @asyncapi/generator 3.5.0 and @asyncapi/markdown-template 2.0.0 are the newest releases, and every fix is a major + # version past a range one of them or their dependencies requires, so no reinstall reaches any of them. They run + # only in `gen:ws-md`, a developer-time render of generated/specs/ws.yaml -- a spec this repo generates from its + # own Rust types -- through the template's local install path, so every input is ours, nothing is fetched or + # cloned, and neither tool is shipped or reachable from a deployment. + # + # @babel/core 7.12.9 reads any source map named by a sourceMappingURL in the code it compiles, fixed in 7.29.6. + # @asyncapi/generator-react-sdk 1.1.3 pins it by equality. It compiles only the template's own JSX. + { id = "GHSA-4x5r-pxfx-6jf8", reason = "expires 2026-11-13" }, + # simple-git 3.36.0 and its @simple-git/argv-parser 1.1.1 have gaps in their unsafe-operations guard. + # It misses `VISUAL`, abbreviated long options, `include.path` and `trailer..cmd`, fixed in simple-git 4, + # and the generator requires ^3.36.0. All four need attacker-controlled git arguments or config, and the generator + # reaches git only to clone a template named by a remote URL, which `gen:ws-md` never passes. + { id = "GHSA-858h-whjf-mvg5", reason = "expires 2026-11-13" }, + { id = "GHSA-g4wm-2vf7-vfgr", reason = "expires 2026-11-13" }, + { id = "GHSA-v5rq-49vh-5v5c", reason = "expires 2026-11-13" }, + { id = "GHSA-x6jw-m9v5-85vh", reason = "expires 2026-11-13" }, + # ajv 5.5.2 and 6.5.2: ReDoS through a `$data` pattern, and prototype pollution from a crafted schema. + # Pulled by json-schema-migrate 0.2.0 and webapi-parser 0.5.0 under the generator's parser stack; both need an + # attacker-supplied schema or instance, and the only schema validated here is our own spec. + { id = "GHSA-2g4f-4pwh-qvx6", reason = "expires 2026-11-13" }, + { id = "GHSA-v88g-cgmw-v5xw", reason = "expires 2026-11-13" }, + # jsonpath-plus 7.2.0 code execution through a crafted JSONPath expression, fixed in 10.3.0. + # Pulled by the parser 1.x and 2.x copies @asyncapi/multi-parser bundles for old spec versions. The expressions + # come from the parser's own rulesets and our spec, never from an untrusted source. + { id = "GHSA-hw8r-x6gr-5gjp", reason = "expires 2026-11-13" }, + { id = "GHSA-pppg-cpfq-h7wr", reason = "expires 2026-11-13" }, + # glob 10.5.0 / 13.0.6 and minimatch 3.1.5 / 9.0.9 / 10.2.6 can be made to do excess work. + # glob keeps walking after its AbortSignal fires, and minimatch amplifies a crafted pattern. Both are fixed only in + # new majors that @npmcli and @stoplight require below. Each costs time only, and only for a caller that aborts a + # walk or matches an attacker's pattern; the patterns here are fixed. + { id = "GHSA-3g3m-2fjr-6pw4", reason = "expires 2026-11-13" }, + { id = "GHSA-g3hv-75wg-m249", reason = "expires 2026-11-13" }, ] [licenses] diff --git a/config/jscpd-baseline.json b/config/jscpd-baseline.json index 9407a0b6..c3553f23 100644 --- a/config/jscpd-baseline.json +++ b/config/jscpd-baseline.json @@ -10,6 +10,7 @@ "03e4e07acff327e2": 1, "04199950b57990a0": 1, "04daef50421ed235": 1, + "05f89321b4dbba3a": 1, "0690b4ae1dc710a4": 1, "07378241fe562817": 1, "075c84a635548077": 1, @@ -42,6 +43,7 @@ "1a1e1611e4db4656": 1, "1a46175d923bf850": 3, "1a97e5c72a30152e": 1, + "1b8894933d283827": 1, "1bd6f5989d1961e5": 1, "1c64414236b8476c": 1, "1c8cf13e10b67513": 1, @@ -161,6 +163,7 @@ "71d4c22b912832b9": 1, "72295cb3aacc3d6c": 1, "726487c2d82f5f69": 1, + "727e7a5304bdc471": 1, "728c97d5724db2c3": 1, "73b4d0b9adadfa1f": 1, "73ffadbd8a93c577": 1, @@ -319,6 +322,7 @@ "dd162080de996936": 1, "dd41701c3663b9f3": 1, "ddca47a4caded0fd": 1, + "de1a3233f93e1695": 1, "dfab26504ac7abe4": 1, "e152738c60f43b0e": 1, "e25456ad9cc7f9f9": 1, @@ -333,6 +337,7 @@ "eb95988e62f801fd": 1, "ebb8f005bcebfcee": 1, "ec5c408a8f2c368d": 1, + "ec83e5f612762cae": 1, "edfb18c73a139873": 1, "ef69568ef92f1b79": 1, "ef6dd6376769473d": 1, diff --git a/config/osv-scanner.toml b/config/osv-scanner.toml index e356856a..7777f9d2 100644 --- a/config/osv-scanner.toml +++ b/config/osv-scanner.toml @@ -1,9 +1,20 @@ # AUTO-GENERATED from config/deny.toml by `mise run gen:osv-scanner`. # The list spans every language, so "has unused ignores" warnings are expected: agents must ignore them. IgnoredVulns = [ + { id = "GHSA-2g4f-4pwh-qvx6", ignoreUntil = 2026-11-13 }, + { id = "GHSA-3g3m-2fjr-6pw4", ignoreUntil = 2026-11-13 }, + { id = "GHSA-4x5r-pxfx-6jf8", ignoreUntil = 2026-11-13 }, { id = "GHSA-5j98-2g5x-46v6", ignoreUntil = 2026-11-13 }, { id = "GHSA-6f2x-v7q7-m7m5", ignoreUntil = 2026-11-13 }, + { id = "GHSA-858h-whjf-mvg5", ignoreUntil = 2026-11-13 }, + { id = "GHSA-g3hv-75wg-m249", ignoreUntil = 2026-11-13 }, + { id = "GHSA-g4wm-2vf7-vfgr", ignoreUntil = 2026-11-13 }, + { id = "GHSA-hw8r-x6gr-5gjp", ignoreUntil = 2026-11-13 }, + { id = "GHSA-pppg-cpfq-h7wr", ignoreUntil = 2026-11-13 }, { id = "GHSA-qwgh-2vcv-g2f7", ignoreUntil = 2026-11-13 }, + { id = "GHSA-v5rq-49vh-5v5c", ignoreUntil = 2026-11-13 }, + { id = "GHSA-v88g-cgmw-v5xw", ignoreUntil = 2026-11-13 }, + { id = "GHSA-x6jw-m9v5-85vh", ignoreUntil = 2026-11-13 }, { id = "RUSTSEC-2023-0071", ignoreUntil = 2026-11-13 }, { id = "RUSTSEC-2024-0436", ignoreUntil = 2026-11-13 }, { id = "RUSTSEC-2024-0449", ignoreUntil = 2026-11-13 }, diff --git a/services/ws-wasi-runner/Cargo.toml b/services/ws-wasi-runner/Cargo.toml index 7c344752..ffad541c 100644 --- a/services/ws-wasi-runner/Cargo.toml +++ b/services/ws-wasi-runner/Cargo.toml @@ -23,18 +23,69 @@ path = "src/lib.rs" doctest = false name = "et-ws-wasi-runner" path = "src/main.rs" +required-features = ["std"] + +# The tests of the `std` runner, and of the board's, which compiles its guests as only a host can. +# `server_message_to_wit`, the test of the conversions a board shares, runs with `--no-default-features` as well. +[[test]] +name = "board" +required-features = ["std"] + +[[test]] +name = "modules" +required-features = ["std"] + +[[test]] +name = "o2_ingest_search" +required-features = ["std"] + +[[test]] +name = "otel_propagation" +required-features = ["std"] + +[[test]] +name = "vector_otlp_relay" +required-features = ["std"] + +[[test]] +name = "ws_backend" +required-features = ["std"] [features] #! Each standardised WASI interface hosted beyond `et:ws-wasi` and WASI Preview 2 is a feature, all on by default. #! A guest that imports an interface the build left out fails to instantiate, naming the import. -default = ["nn", "webgpu"] +default = ["nn", "std", "webgpu"] +## The runner for a host with an operating system. +## It fetches a component from the hub, compiles it, and hosts WASI Preview 2 through `wasmtime-wasi`. Without it the +## crate is `no_std`, and runs a component compiled ahead of time for a board through its `board` module. +std = [ + "chrono/std", + "dep:et-otlp", + "dep:et-rest-client", + "dep:et-ws-runner-common", + "dep:fs-err", + "dep:futures-util", + "dep:libc", + "dep:reqwest", + "dep:tokio", + "dep:tokio-tungstenite", + "dep:tracing", + "dep:wasmtime-wasi", + "edge-toolkit/std", + "precompile", + "serde_json/std", + "thiserror/std", + "wasmtime/default", +] +## Compiles a component ahead of time for a board's engine, as a firmware's build script does. +precompile = ["wasmtime/cranelift", "wasmtime/std"] ## Hosts `wasi:nn` (graph, tensor, inference) on ONNX Runtime through `wasmtime-wasi-nn`. ## Guests can then load an ONNX model and run inference on it. -nn = ["dep:ort", "dep:wasmtime-wasi-nn"] +nn = ["dep:ort", "dep:wasmtime-wasi-nn", "std"] ## Hosts the full upstream `wasi:webgpu` interface through `wasi-webgpu-wasmtime`. ## A wgpu instance on every native backend gives guests real GPU compute and rendering. -webgpu = ["dep:wasi-webgpu-wasmtime", "dep:wgpu-core", "dep:wgpu-types"] +webgpu = ["dep:wasi-webgpu-wasmtime", "dep:wgpu-core", "dep:wgpu-types", "std"] ## Dispatches `wasi:nn` inference on `ExecutionTarget::Gpu` to ONNX Runtime's `CUDAExecutionProvider`. ## Without it such a request warns and falls back to the CPU. It enables wasmtime-wasi-nn's `onnx-cuda`, and ## through it `ort/cuda`, so `ort/download-binaries` pulls the CUDA-flavoured ONNX Runtime prebuilt -- which only @@ -44,32 +95,34 @@ cuda = ["nn", "wasmtime-wasi-nn/onnx-cuda"] ## Internal: compiles in the `/cov` preopen for the repo's coverage workflow. ## Mapped to `target/wasi-cov`, it is where instrumented WASI guests dump their minicov `.profraw`. ## `ET_TEST_COVERAGE` still gates the preopen at runtime. -coverage = ["et-ws-runner-common/coverage"] +coverage = ["et-ws-runner-common?/coverage", "std"] ## Internal: renders this feature list into the crate documentation, as docs.rs builds it. docs = ["dep:document-features"] +# What a board's `no_std` runner needs as much as the host's: the engine, and the message types and their JSON. [dependencies] -chrono = { workspace = true, features = ["std"] } +chrono = { workspace = true, features = ["alloc"] } document-features = { workspace = true, optional = true } -edge-toolkit = { workspace = true, features = ["std"] } -et-otlp.workspace = true -et-rest-client = { workspace = true, features = ["tracing"] } -et-ws-runner-common.workspace = true -fs-err.workspace = true -futures-util.workspace = true +edge-toolkit.workspace = true +serde_json = { workspace = true, features = ["alloc"] } +serde_path_to_error.workspace = true +thiserror.workspace = true +wasmtime.workspace = true +# The rest is the `std` runner's. +et-otlp = { workspace = true, optional = true } +et-rest-client = { workspace = true, features = ["tracing"], optional = true } +et-ws-runner-common = { workspace = true, optional = true } +fs-err = { workspace = true, optional = true } +futures-util = { workspace = true, optional = true } # libc::_exit for the macOS test-only fast exit (see main.rs). # std::process::exit still runs atexit handlers, which is exactly the path that races libc++ during # ORT teardown -- _exit skips them. -libc.workspace = true -reqwest = { workspace = true, features = ["json", "rustls", "stream"] } -serde_json = { workspace = true, features = ["std"] } -serde_path_to_error.workspace = true -thiserror = { workspace = true, features = ["std"] } -tokio = { workspace = true, features = ["macros", "net", "rt-multi-thread", "sync", "time"] } -tokio-tungstenite = { workspace = true, features = ["connect"] } -tracing = { workspace = true, features = ["std"] } -wasmtime.workspace = true -wasmtime-wasi.workspace = true +libc = { workspace = true, optional = true } +reqwest = { workspace = true, features = ["json", "rustls", "stream"], optional = true } +tokio = { workspace = true, features = ["macros", "net", "rt-multi-thread", "sync", "time"], optional = true } +tokio-tungstenite = { workspace = true, features = ["connect"], optional = true } +tracing = { workspace = true, features = ["std"], optional = true } +wasmtime-wasi = { workspace = true, optional = true } # wasi-nn standardised ML inference. # The workspace pin drops `openvino` + `winml` (which we don't use) and keeps `onnx`. wasmtime-wasi-nn = { workspace = true, optional = true } diff --git a/services/ws-wasi-runner/README.md b/services/ws-wasi-runner/README.md index 0766aeb7..c2746562 100644 --- a/services/ws-wasi-runner/README.md +++ b/services/ws-wasi-runner/README.md @@ -37,3 +37,19 @@ Plus, attached to the same `Linker` but defined by external WIT packages, each b wasmtime-wasi-nn 48 still uses). Guests load model bytes via `graph.load`, build `Tensor`s, and call `compute` -- the same shape of calls Spin / wasmCloud / Fermyon production wasi-nn workloads use. Inference runs on the CPU unless the `cuda` feature is on. + +## On a board + +Without its default `std` feature the crate is `no_std` + `alloc`, and its `board` module runs a guest on a +microcontroller with no operating system. A board cannot compile WebAssembly, so its engine runs Pulley bytecode +instead: a firmware's build script compiles the guest ahead of time with `engine::precompile` (the `precompile` +feature) for the board's Pulley target, `pulley32` on a 32-bit board, and builds the result in. + +`board::run` hosts the `runner` world on what the firmware provides through two traits: a `Platform`, the board's +clock and log, and a `Transport`, its websocket to the hub. Guest buckets live in the board's memory. Of WASI +Preview 2 it hosts only what the ws-module guests import -- their standard streams, an empty environment, the +monotonic clock and polling -- since `wasmtime-wasi` needs `std`; `wasi:nn` and `wasi:webgpu` are host-only. + +wasmtime asks the firmware for one thing more, a thread-local pointer through `wasmtime_tls_get` and +`wasmtime_tls_set`, which on a single core is a static. A guest's linear memory is its declared size, about 1 MiB +for the Rust guests, so a board needs external RAM such as PSRAM to hold it. diff --git a/services/ws-wasi-runner/src/bindings.rs b/services/ws-wasi-runner/src/bindings.rs index 109a7f47..9e49c89e 100644 --- a/services/ws-wasi-runner/src/bindings.rs +++ b/services/ws-wasi-runner/src/bindings.rs @@ -18,7 +18,7 @@ A wasmtime bump that emits a fifth fails clippy until this list grows, which is the loud direction." )] #[doc(hidden)] -pub use super::host::wasi_keyvalue::Bucket as __with_name0; +pub use super::keyvalue::Bucket as __with_name0; /// Auto-generated bindings for a pre-instantiated version of a /// component which implements the world `runner`. /// diff --git a/services/ws-wasi-runner/src/board/cov.rs b/services/ws-wasi-runner/src/board/cov.rs new file mode 100644 index 00000000..d63b7ed6 --- /dev/null +++ b/services/ws-wasi-runner/src/board/cov.rs @@ -0,0 +1,116 @@ +//! The `/cov` directory a guest built for coverage writes its profile into as its `run` ends. +//! +//! An instrumented guest imports `wasi:filesystem` for that one write, and could not instantiate on a board without +//! it. What it writes is discarded: the host runner runs the same guests and collects their profiles, so the board's +//! run would only replace one. The guest's libc imports more of a descriptor than the write needs, and those +//! functions answer `unsupported`. Each is defined dynamically, so that the guest's types describe their values +//! rather than a copy of every flag and error code. + +use wasmtime::component::types::ComponentFunc; +use wasmtime::component::{Linker, Resource, ResourceType, Val}; + +use super::host::BoardHost; +use super::platform::{Platform, Transport}; +use super::wasi::{Ctx, OutputStream, PREVIEW2, drop_from_table}; + +/// A `wasi:filesystem` descriptor: `/cov`, or a file the guest opened in it. +#[derive(Clone, Copy)] +enum Descriptor { + /// `/cov` itself. + Directory, + /// A file in `/cov`. + File, +} + +/// Define `wasi:filesystem`'s `types` and `preopens`, with `/cov` the only directory. +/// +/// # Errors +/// +/// When the linker already holds one of them. +#[expect(clippy::single_call_fn, reason = "the coverage step of linking WASI")] +pub(super) fn add_to_linker(linker: &mut Linker>) -> wasmtime::Result<()> +where + P: Platform + 'static, + T: Transport + 'static, +{ + let mut types = linker.instance(&format!("wasi:filesystem/types@{PREVIEW2}"))?; + types.resource( + "descriptor", + ResourceType::host::(), + drop_from_table::, + )?; + types.func_new( + "[method]descriptor.open-at", + |mut store: Ctx<'_, P, T>, _: ComponentFunc, params: &[Val], results: &mut [Val]| { + let outcome = match descriptor(&mut store, params)? { + Descriptor::Directory => { + let file = store.data_mut().table().push(Descriptor::File)?; + Ok(Val::Resource(file.try_into_resource_any(&mut store)?)) + } + Descriptor::File => Err("not-directory"), + }; + answer(results, outcome) + }, + )?; + types.func_new( + "[method]descriptor.write-via-stream", + |mut store: Ctx<'_, P, T>, _: ComponentFunc, params: &[Val], results: &mut [Val]| { + let outcome = match descriptor(&mut store, params)? { + Descriptor::File => { + let stream = store.data_mut().table().push(OutputStream::Discard)?; + Ok(Val::Resource(stream.try_into_resource_any(&mut store)?)) + } + Descriptor::Directory => Err("is-directory"), + }; + answer(results, outcome) + }, + )?; + for method in [ + "read-via-stream", + "append-via-stream", + "get-flags", + "stat", + "metadata-hash", + ] { + types.func_new( + &format!("[method]descriptor.{method}"), + |_: Ctx<'_, P, T>, _: ComponentFunc, _: &[Val], results: &mut [Val]| answer(results, Err("unsupported")), + )?; + } + + linker + .instance(&format!("wasi:filesystem/preopens@{PREVIEW2}"))? + .func_wrap("get-directories", |mut store: Ctx<'_, P, T>, (): ()| { + let directory = store.data_mut().table().push(Descriptor::Directory)?; + Ok((vec![(directory, "/cov".to_owned())],)) + }) +} + +/// The descriptor a `[method]descriptor.*` call was made on, its first parameter. +fn descriptor(store: &mut Ctx<'_, P, T>, params: &[Val]) -> wasmtime::Result +where + P: 'static, + T: 'static, +{ + let Some(Val::Resource(this)) = params.first() else { + return Err(wasmtime::Error::msg( + "a descriptor method was called without its descriptor", + )); + }; + let this: Resource = this.try_into_resource(&mut *store)?; + Ok(*store.data_mut().table().get(&this)?) +} + +/// Return `outcome` as the `result<_, error-code>` every descriptor function returns, its error named by its case. +fn answer(results: &mut [Val], outcome: Result) -> wasmtime::Result<()> { + let Some(result) = results.first_mut() else { + return Err(wasmtime::Error::msg( + "a descriptor function was called without room for its result", + )); + }; + *result = Val::Result(match outcome { + Ok(value) => Ok(Some(Box::new(value))), + Err(code) => Err(Some(Box::new(Val::Enum(code.to_owned())))), + }); + Ok(()) +} diff --git a/services/ws-wasi-runner/src/board/host.rs b/services/ws-wasi-runner/src/board/host.rs new file mode 100644 index 00000000..24003ded --- /dev/null +++ b/services/ws-wasi-runner/src/board/host.rs @@ -0,0 +1,237 @@ +//! A board's store data, and its hosting of the `runner` world: the guest's log through the board, its buckets in +//! the board's memory, and its websocket over the board's transport. + +use alloc::collections::{BTreeMap, VecDeque}; +#[cfg(not(feature = "std"))] +use alloc::{ + string::{String, ToString as _}, + vec::Vec, +}; + +use edge_toolkit::ws::{ClientMessage, ServerMessage}; +use wasmtime::component::{Resource, ResourceTable}; + +use super::platform::{Platform, Transport}; +use crate::bindings::et::ws_messages::messages::{ + ClientMessage as WitClientMessage, ServerMessage as WitServerMessage, +}; +use crate::bindings::et::ws_wasi::ws::{State, WsError}; +use crate::bindings::wasi::keyvalue::store::{Error as StoreError, KeyResponse}; +use crate::bindings::wasi::logging::logging::Level; +use crate::bindings::{et, wasi}; +use crate::convert::{Frame, client_frame, server_message, server_message_to_wit}; +use crate::keyvalue::{Bucket, bucket_from_identifier, kv_not_implemented}; + +/// How long `connect` waits for each frame while it waits for the hub's `et-connect-ack`. +const CONNECT_ACK_TIMEOUT_MS: u32 = 10_000; + +/// The store data a board runs a guest with. +pub struct BoardHost { + /// The board's clock, log and console. + platform: P, + /// The guest's resources: its buckets, streams and pollables. + table: ResourceTable, + /// The websocket to the hub. + transport: T, + /// Every storage bucket the guest has written, by its agent id, each a map from key to value. + buckets: BTreeMap>>, + /// Messages that arrived before the guest asked for them: the connect ack, and any that came ahead of it. + inbox: VecDeque, + /// The id the hub assigned, empty until it has. + agent_id: String, + /// The websocket's state, as the guest sees it. + state: State, +} + +impl BoardHost { + /// A guest's store data, with nothing yet stored and no websocket open. + #[cfg_attr( + not(test), + expect( + clippy::single_call_fn, + reason = "the store data's constructor, called where run makes the store" + ) + )] + pub(super) fn new(platform: P, transport: T) -> Self { + Self { + platform, + table: ResourceTable::new(), + transport, + buckets: BTreeMap::new(), + inbox: VecDeque::new(), + agent_id: String::default(), + state: State::Closed, + } + } + + /// The board's clock, log and console, which the WASI functions use. + pub(super) const fn platform(&self) -> &P { + &self.platform + } + + /// The guest's resources, which the WASI functions add their streams and pollables to. + pub(super) const fn table(&mut self) -> &mut ResourceTable { + &mut self.table + } + + /// Whether the websocket is open, as every call but `connect` needs it. + const fn connected(&self) -> Result<(), WsError> { + if matches!(self.state, State::Connected) { + Ok(()) + } else { + Err(WsError::NotConnected) + } + } +} + +impl BoardHost { + /// Send the hello over the open transport and wait for the hub's ack, queueing it and anything ahead of it. + async fn register(&mut self) -> Result<(), WsError> { + let hello = serde_json::to_string(&ClientMessage::Connect { agent_id: None })?; + if let Err(reason) = self.transport.send(Frame::Text(hello)).await { + return Err(WsError::Transport(reason.to_string())); + } + loop { + let frame = match self.transport.recv(CONNECT_ACK_TIMEOUT_MS).await { + Ok(Some(frame)) => frame, + Ok(None) => return Err(WsError::Transport("the hub sent no et-connect-ack".to_string())), + Err(reason) => return Err(WsError::Transport(reason.to_string())), + }; + let message = server_message(frame)?; + if let ServerMessage::ConnectAck { agent_id, .. } = &message { + self.agent_id.clone_from(agent_id); + self.inbox.push_front(message); + self.state = State::Connected; + return Ok(()); + } + self.inbox.push_back(message); + } + } +} + +impl wasi::logging::logging::Host for BoardHost { + async fn log(&mut self, level: Level, context: String, message: String) { + self.platform.log(level, &context, &message); + } +} + +#[expect( + clippy::unused_async_trait_impl, + reason = "the generated Host trait declares open async, and the buckets are in memory" +)] +impl wasi::keyvalue::store::Host for BoardHost { + async fn open(&mut self, identifier: String) -> Result, StoreError> { + let bucket = bucket_from_identifier(&identifier)?; + Ok(self.table.push(bucket)?) + } +} + +#[expect( + clippy::unused_async_trait_impl, + reason = "the generated HostBucket trait declares every method async, and the buckets are in memory" +)] +impl wasi::keyvalue::store::HostBucket for BoardHost { + async fn get(&mut self, self_: Resource, key: String) -> Result>, StoreError> { + Ok(match self.table.get(&self_)? { + Bucket::Storage { agent_id } => self.buckets.get(agent_id).and_then(|stored| stored.get(&key)).cloned(), + // A board serves no module's files, so none of them holds a key. + Bucket::Modules { .. } => None, + }) + } + + async fn set(&mut self, self_: Resource, key: String, value: Vec) -> Result<(), StoreError> { + let Bucket::Storage { agent_id } = self.table.get(&self_)? else { + return Err(StoreError::AccessDenied); + }; + let _previous = self.buckets.entry(agent_id.clone()).or_default().insert(key, value); + Ok(()) + } + + async fn delete(&mut self, _bucket: Resource, _key: String) -> Result<(), StoreError> { + Err(kv_not_implemented("delete")) + } + + async fn exists(&mut self, _bucket: Resource, _key: String) -> Result { + Err(kv_not_implemented("exists")) + } + + async fn list_keys( + &mut self, + _bucket: Resource, + _cursor: Option, + ) -> Result { + Err(kv_not_implemented("list-keys")) + } + + async fn drop(&mut self, rep: Resource) -> wasmtime::Result<()> { + let _dropped: Bucket = self.table.delete(rep)?; + Ok(()) + } +} + +// The `et:ws-messages/messages` interface only declares types, but the linker still anchors it on a `Host` impl. +impl et::ws_messages::messages::Host for BoardHost {} + +impl et::ws_wasi::ws::Host for BoardHost { + /// Open the websocket and register, keeping the hub's ack for the guest's first `recv`, as the host runner does. + async fn connect(&mut self) -> Result<(), WsError> { + if matches!(self.state, State::Connected) { + return Err(WsError::AlreadyConnected); + } + if let Err(reason) = self.transport.connect().await { + return Err(WsError::Transport(reason.to_string())); + } + self.state = State::Connecting; + let registered = self.register().await; + if registered.is_err() { + et::ws_wasi::ws::Host::disconnect(self).await; + } + registered + } + + #[expect( + clippy::unused_async_trait_impl, + reason = "the generated Host trait declares get-state async, and the state is a field" + )] + async fn get_state(&mut self) -> State { + self.state + } + + #[expect( + clippy::unused_async_trait_impl, + reason = "the generated Host trait declares agent-id async, and the id is a field" + )] + async fn agent_id(&mut self) -> String { + self.agent_id.clone() + } + + async fn send(&mut self, message: WitClientMessage) -> Result<(), WsError> { + self.connected()?; + match self.transport.send(client_frame(message)?).await { + Ok(()) => Ok(()), + Err(reason) => Err(WsError::Transport(reason.to_string())), + } + } + + async fn recv(&mut self, timeout_ms: u32) -> Result, WsError> { + self.connected()?; + let message = match self.inbox.pop_front() { + Some(message) => message, + None => match self.transport.recv(timeout_ms).await { + Ok(Some(frame)) => server_message(frame)?, + Ok(None) => return Ok(None), + Err(reason) => return Err(WsError::Transport(reason.to_string())), + }, + }; + Ok(Some(server_message_to_wit(message)?)) + } + + async fn disconnect(&mut self) { + if !matches!(self.state, State::Closed) { + self.transport.close().await; + } + self.state = State::Closed; + self.inbox.clear(); + self.agent_id.clear(); + } +} diff --git a/services/ws-wasi-runner/src/board/mod.rs b/services/ws-wasi-runner/src/board/mod.rs new file mode 100644 index 00000000..09a33537 --- /dev/null +++ b/services/ws-wasi-runner/src/board/mod.rs @@ -0,0 +1,17 @@ +//! The runner for a board with no operating system. +//! +//! It runs a component compiled ahead of time for the board's engine, with the `runner` world and the WASI Preview 2 +//! its guest imports hosted on what the board provides. + +#[cfg(feature = "coverage")] +mod cov; +mod host; +mod platform; +mod run; +#[cfg(test)] +mod test_host; +mod wasi; + +pub use self::host::BoardHost; +pub use self::platform::{Console, Platform, Transport}; +pub use self::run::{BoardError, run}; diff --git a/services/ws-wasi-runner/src/board/platform.rs b/services/ws-wasi-runner/src/board/platform.rs new file mode 100644 index 00000000..f6a2b861 --- /dev/null +++ b/services/ws-wasi-runner/src/board/platform.rs @@ -0,0 +1,64 @@ +//! What a board gives its runner: a clock and a log, and a websocket to the hub. + +use crate::bindings::wasi::logging::logging::Level; +use crate::convert::Frame; + +/// Which of the guest's standard streams a write went to. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +#[non_exhaustive] +pub enum Console { + /// `wasi:cli/stdout`. + Stdout, + /// `wasi:cli/stderr`. + Stderr, +} + +/// The board's clock, and where the guest's log and console go, shared by every host call the guest makes. +pub trait Platform: Send + Sync { + /// Nanoseconds since some fixed point, which only ever increase. + fn now(&self) -> u64; + + /// Wait until [`Platform::now`] reaches `deadline`. + fn sleep_until(&self, deadline: u64) -> impl Future + Send; + + /// Record one `wasi:logging` entry from the guest. + fn log(&self, level: Level, context: &str, message: &str); + + /// Write what the guest wrote to one of its standard streams. + fn write(&self, console: Console, bytes: &[u8]); +} + +/// A websocket to the hub, opened on the guest's `connect`. +/// +/// Keeping the connection alive is the transport's: the hub closes one it has heard nothing on for 15 seconds, so a +/// transport pings it while the guest is busy elsewhere. +pub trait Transport: Send { + /// Why the websocket failed, which the guest is told as a `transport` error. + type Error: core::fmt::Display; + + /// Open the websocket. + /// + /// # Errors + /// + /// When the websocket could not be opened. + fn connect(&mut self) -> impl Future> + Send; + + /// Send one data frame. + /// + /// # Errors + /// + /// When the frame could not be sent. + fn send(&mut self, frame: Frame) -> impl Future> + Send; + + /// The next data frame to arrive within `timeout_ms`. + /// + /// `None` means none arrived in that time, which is not a failure: it is what `recv` tells the guest then. + /// + /// # Errors + /// + /// When the websocket could not be read. + fn recv(&mut self, timeout_ms: u32) -> impl Future, Self::Error>> + Send; + + /// Close the websocket. + fn close(&mut self) -> impl Future + Send; +} diff --git a/services/ws-wasi-runner/src/board/run.rs b/services/ws-wasi-runner/src/board/run.rs new file mode 100644 index 00000000..97b01481 --- /dev/null +++ b/services/ws-wasi-runner/src/board/run.rs @@ -0,0 +1,64 @@ +//! Run a component compiled ahead of time for a board, with the `runner` world and WASI hosted by the board. + +use wasmtime::component::{Component, HasSelf, Linker}; +use wasmtime::{Engine, Store}; + +use super::host::BoardHost; +use super::platform::{Platform, Transport}; +use super::wasi; +use crate::bindings; +use crate::bindings::exports::et::ws_wasi::entry::EntryError; + +/// Why a board's guest did not run to completion. +#[derive(Debug, thiserror::Error)] +#[non_exhaustive] +pub enum BoardError { + /// The component could not be loaded, linked or instantiated, or trapped while it ran. + #[error("{0}")] + Wasmtime(wasmtime::Error), + /// The guest's `entry.run` returned an error. + #[error("the guest failed: {0:?}")] + Guest(EntryError), +} + +impl From for BoardError { + fn from(err: wasmtime::Error) -> Self { + Self::Wasmtime(err) + } +} + +/// Run `component` on `engine` until its `entry.run` returns, hosting it on `platform` and `transport`. +/// +/// `engine` is [`crate::engine::board_engine`]'s, and `component` what [`crate::engine::precompile`] compiled for +/// it: a firmware builds its guest in, compiled by its build script. +/// +/// # Errors +/// +/// When `component` was not compiled for `engine`, the guest imports something a board does not host, it traps, +/// or its `entry.run` returns an error. +pub async fn run(engine: &Engine, component: &[u8], platform: P, transport: T) -> Result<(), BoardError> +where + P: Platform + 'static, + T: Transport + 'static, +{ + #[expect( + unsafe_code, + reason = "a precompiled component is native code to wasmtime, so loading one is unsafe by nature" + )] + // SAFETY: the bytes are what `precompile` produced for this engine's settings, built into the firmware by its + // build script rather than received from anywhere at run time. + let component = unsafe { Component::deserialize(engine, component) }?; + let mut linker = Linker::new(engine); + wasi::add_to_linker(&mut linker)?; + bindings::Runner::add_to_linker::, HasSelf>>(&mut linker, |host| host)?; + let mut store = Store::new(engine, BoardHost::new(platform, transport)); + let runner = bindings::Runner::instantiate_async(&mut store, &component, &linker).await?; + // `entry.run` is an async export, which only the concurrent API can call. + match store + .run_concurrent(async |accessor| runner.et_ws_wasi_entry().call_run(accessor).await) + .await?? + { + Ok(()) => Ok(()), + Err(error) => Err(BoardError::Guest(error)), + } +} diff --git a/services/ws-wasi-runner/src/board/test_host.rs b/services/ws-wasi-runner/src/board/test_host.rs new file mode 100644 index 00000000..c9eef57b --- /dev/null +++ b/services/ws-wasi-runner/src/board/test_host.rs @@ -0,0 +1,152 @@ +//! Covers the `runner` world calls no shipped guest makes on a board: the websocket used out of order, and the +//! bucket calls a board refuses. +#![cfg(test)] + +#[cfg(not(feature = "std"))] +use alloc::{borrow::ToOwned as _, string::String, vec, vec::Vec}; +use core::pin::pin; +use core::task::{Context, Poll, Waker}; + +use edge_toolkit::ws::{ConnectStatus, ServerMessage}; +use wasmtime::component::Resource; + +use super::host::BoardHost; +use super::platform::{Console, Platform, Transport}; +use crate::bindings::et::ws_messages::messages::ClientMessage; +use crate::bindings::et::ws_wasi::ws::{Host as _, State, WsError}; +use crate::bindings::wasi::keyvalue::store::{Error, Host as _, HostBucket}; +use crate::bindings::wasi::logging::logging::Level; +use crate::convert::Frame; +use crate::keyvalue::Bucket; + +/// A board with nothing to say: its clock stands still and its log goes nowhere. +struct Quiet; + +impl Platform for Quiet { + fn now(&self) -> u64 { + 0 + } + + async fn sleep_until(&self, _deadline: u64) {} + + fn log(&self, _level: Level, _context: &str, _message: &str) {} + + fn write(&self, _console: Console, _bytes: &[u8]) {} +} + +/// A hub that acks every connect, and has nothing else to say. +struct Acking; + +#[expect( + clippy::unused_async_trait_impl, + reason = "the hub answers at once from memory, where a board's transport awaits its socket" +)] +impl Transport for Acking { + type Error = &'static str; + + async fn connect(&mut self) -> Result<(), &'static str> { + Ok(()) + } + + async fn send(&mut self, _frame: Frame) -> Result<(), &'static str> { + Ok(()) + } + + async fn recv(&mut self, _timeout_ms: u32) -> Result, &'static str> { + let ack = ServerMessage::ConnectAck { + agent_id: "agent-1".to_owned(), + status: ConnectStatus::Assigned, + }; + Ok(Some(Frame::Text(serde_json::to_string(&ack).unwrap()))) + } + + async fn close(&mut self) {} +} + +/// `future`'s output, which every call below has at once, since nothing in them waits. +fn now(future: F) -> F::Output +where + F: Future, +{ + match pin!(future).poll(&mut Context::from_waker(Waker::noop())) { + Poll::Ready(output) => output, + Poll::Pending => panic!("a board call waited on nothing"), + } +} + +/// A guest's store data on a quiet board with an acking hub. +fn host() -> BoardHost { + BoardHost::new(Quiet, Acking) +} + +/// The bucket `identifier` opens. +fn open(host: &mut BoardHost, identifier: &str) -> Resource { + now(host.open(identifier.to_owned())).unwrap() +} + +#[test] +fn the_websocket_refuses_messages_until_it_connects() { + let mut host = host(); + assert!(matches!(now(host.get_state()), State::Closed)); + assert!(matches!( + now(host.send(ClientMessage::ListAgents)), + Err(WsError::NotConnected) + )); + assert!(matches!(now(host.recv(0)), Err(WsError::NotConnected))); + now(host.disconnect()); + assert!(matches!(now(host.get_state()), State::Closed)); +} + +#[test] +fn a_connected_websocket_refuses_a_second_connect() { + let mut host = host(); + now(host.connect()).unwrap(); + assert!(matches!(now(host.connect()), Err(WsError::AlreadyConnected))); + assert!(matches!(now(host.get_state()), State::Connected)); + assert_eq!(now(host.agent_id()), "agent-1"); + now(host.disconnect()); + assert!(matches!(now(host.get_state()), State::Closed)); + assert_eq!(now(host.agent_id()), String::new()); +} + +#[test] +fn a_module_bucket_holds_nothing_and_cannot_be_written() { + let mut host = host(); + let bucket = open(&mut host, "modules/wasi-math1"); + let rep = bucket.rep(); + assert!(matches!( + now(host.get(Resource::new_own(rep), "pkg/model.onnx".to_owned())), + Ok(None) + )); + assert!(matches!( + now(host.set(Resource::new_own(rep), "pkg/model.onnx".to_owned(), vec![1])), + Err(Error::AccessDenied) + )); + now(HostBucket::drop(&mut host, bucket)).unwrap(); +} + +#[test] +fn a_bucket_refuses_the_calls_a_board_does_not_implement() { + let mut host = host(); + let rep = open(&mut host, "agent-1").rep(); + let refusals = [ + now(host.delete(Resource::new_own(rep), "key".to_owned())).err(), + now(host.exists(Resource::new_own(rep), "key".to_owned())).err(), + now(host.list_keys(Resource::new_own(rep), None)).err(), + ]; + let details: Vec<_> = refusals + .into_iter() + .map(|refusal| match refusal { + Some(Error::Other(detail)) => detail, + _ => panic!("a bucket call a board does not implement did not refuse as other"), + }) + .collect(); + assert_eq!( + details, + [ + "delete not implemented", + "exists not implemented", + "list-keys not implemented" + ] + ); +} diff --git a/services/ws-wasi-runner/src/board/wasi.rs b/services/ws-wasi-runner/src/board/wasi.rs new file mode 100644 index 00000000..e1257147 --- /dev/null +++ b/services/ws-wasi-runner/src/board/wasi.rs @@ -0,0 +1,283 @@ +//! The WASI Preview 2 a board hosts: what a guest built for `wasm32-wasip2` imports even when it never uses it, +//! defined by hand, since `wasmtime-wasi` needs `std`. +//! +//! Every ws-module guest imports the same few functions: its standard streams and environment, `exit`, the terminal +//! queries, and the monotonic clock and polling it sleeps with. A component imports only the functions it calls, so +//! these are all a guest needs; one importing any other fails to instantiate, naming it. Its standard streams go to +//! the board's console, its environment is empty, it has no terminal, and a pollable is ready at once or at a +//! deadline on the board's clock. +#![expect( + clippy::integer_division_remainder_used, + reason = "wasmtime's ComponentType derive, on StreamError, aligns with `%` in the impls it generates" +)] + +#[cfg(not(feature = "std"))] +use alloc::{boxed::Box, format, string::String, vec::Vec}; + +use wasmtime::StoreContextMut; +use wasmtime::component::{ComponentType, Lift, Linker, Lower, Resource, ResourceType}; + +use super::host::BoardHost; +use super::platform::{Console, Platform, Transport}; + +/// The WASI Preview 2 release the guests import, which any later 0.2 import also resolves to. +pub(super) const PREVIEW2: &str = "0.2.12"; +/// How many bytes `check-write` lets the guest write at once; the board's console takes any amount. +const WRITE_BUDGET: u64 = 4096; + +/// A `wasi:io/poll` pollable: ready now, or at a deadline on the board's clock. +#[derive(Clone, Copy, Debug)] +pub(super) enum Pollable { + /// Ready already, as an output stream always is. + Ready, + /// Ready once [`Platform::now`] reaches this. + Due(u64), +} + +/// A `wasi:io/streams` output stream: where what the guest writes to it goes. +#[derive(Clone, Copy)] +pub(super) enum OutputStream { + /// One of the board's console streams. + Console(Console), + /// Nowhere: a file in the coverage build's `/cov`. + #[cfg(feature = "coverage")] + Discard, +} + +/// A `wasi:io/streams` input stream, the guest's stdin, which never has anything to read. +pub(super) struct InputStream; + +/// A `wasi:io/error`, which a board never raises. +pub(super) enum IoError {} + +/// A `wasi:cli/terminal-input`, which a board never has. +pub(super) enum TerminalInput {} + +/// A `wasi:cli/terminal-output`, which a board never has. +pub(super) enum TerminalOutput {} + +/// `wasi:io/streams`' `stream-error`, which a board's console never fails with. +#[derive(ComponentType, Lift, Lower)] +#[component(variant)] +enum StreamError { + #[component(name = "last-operation-failed")] + LastOperationFailed(Resource), + #[component(name = "closed")] + Closed, +} + +/// The store a host function runs against. +pub(super) type Ctx<'store, P, T> = StoreContextMut<'store, BoardHost>; + +/// Define the WASI Preview 2 functions and resources every ws-module guest imports. +/// +/// # Errors +/// +/// When the linker already holds one of them. +#[expect( + clippy::single_call_fn, + reason = "the WASI step of linking, apart from the runner world's" +)] +pub(super) fn add_to_linker(linker: &mut Linker>) -> wasmtime::Result<()> +where + P: Platform + 'static, + T: Transport + 'static, +{ + add_poll(linker)?; + add_streams(linker)?; + add_cli(linker)?; + #[cfg(feature = "coverage")] + super::cov::add_to_linker(linker)?; + Ok(()) +} + +/// `wasi:io/poll`, and the monotonic clock the guest's pollables come from. +#[expect(clippy::single_call_fn, reason = "the polling part of add_to_linker")] +fn add_poll(linker: &mut Linker>) -> wasmtime::Result<()> +where + P: Platform + 'static, + T: Transport + 'static, +{ + let mut poll = linker.instance(&format!("wasi:io/poll@{PREVIEW2}"))?; + poll.resource( + "pollable", + ResourceType::host::(), + drop_from_table::, + )?; + poll.func_wrap_async( + "[method]pollable.block", + |mut store: Ctx<'_, P, T>, (pollable,): (Resource,)| { + Box::new(async move { + let host = store.data_mut(); + let pollable = *host.table().get(&pollable)?; + let _ready = ready(host.platform(), &[pollable]).await; + Ok(()) + }) + }, + )?; + poll.func_wrap_async( + "poll", + |mut store: Ctx<'_, P, T>, (pollables,): (Vec>,)| { + Box::new(async move { + let host = store.data_mut(); + let mut waiting = Vec::with_capacity(pollables.len()); + for pollable in &pollables { + waiting.push(*host.table().get(pollable)?); + } + Ok((ready(host.platform(), &waiting).await,)) + }) + }, + )?; + + linker + .instance(&format!("wasi:clocks/monotonic-clock@{PREVIEW2}"))? + .func_wrap("subscribe-duration", |mut store: Ctx<'_, P, T>, (duration,): (u64,)| { + let host = store.data_mut(); + let deadline = host.platform().now().saturating_add(duration); + Ok((host.table().push(Pollable::Due(deadline))?,)) + }) +} + +/// `wasi:io/streams` and its errors: the guest's standard streams, written to the board's console. +#[expect(clippy::single_call_fn, reason = "the streams part of add_to_linker")] +fn add_streams(linker: &mut Linker>) -> wasmtime::Result<()> +where + P: Platform + 'static, + T: Transport + 'static, +{ + linker.instance(&format!("wasi:io/error@{PREVIEW2}"))?.resource( + "error", + ResourceType::host::(), + |_, _| Ok(()), + )?; + + let mut streams = linker.instance(&format!("wasi:io/streams@{PREVIEW2}"))?; + streams.resource( + "input-stream", + ResourceType::host::(), + drop_from_table::, + )?; + streams.resource( + "output-stream", + ResourceType::host::(), + drop_from_table::, + )?; + streams.func_wrap( + "[method]output-stream.check-write", + |_: Ctx<'_, P, T>, (_stream,): (Resource,)| Ok((Ok::(WRITE_BUDGET),)), + )?; + streams.func_wrap( + "[method]output-stream.write", + |mut store: Ctx<'_, P, T>, (stream, contents): (Resource, Vec)| { + let host = store.data_mut(); + match *host.table().get(&stream)? { + OutputStream::Console(console) => host.platform().write(console, &contents), + #[cfg(feature = "coverage")] + OutputStream::Discard => {} + } + Ok((Ok::<(), StreamError>(()),)) + }, + )?; + streams.func_wrap( + "[method]output-stream.blocking-flush", + |_: Ctx<'_, P, T>, (_stream,): (Resource,)| Ok((Ok::<(), StreamError>(()),)), + )?; + streams.func_wrap( + "[method]output-stream.subscribe", + |mut store: Ctx<'_, P, T>, (_stream,): (Resource,)| { + Ok((store.data_mut().table().push(Pollable::Ready)?,)) + }, + ) +} + +/// `wasi:cli`: the standard streams, the environment, `exit`, and the terminal queries. +#[expect(clippy::single_call_fn, reason = "the cli part of add_to_linker")] +fn add_cli(linker: &mut Linker>) -> wasmtime::Result<()> +where + P: Platform + 'static, + T: Transport + 'static, +{ + for (interface, getter, console) in [ + ("stdout", "get-stdout", Console::Stdout), + ("stderr", "get-stderr", Console::Stderr), + ] { + linker + .instance(&format!("wasi:cli/{interface}@{PREVIEW2}"))? + .func_wrap(getter, move |mut store: Ctx<'_, P, T>, (): ()| { + Ok((store.data_mut().table().push(OutputStream::Console(console))?,)) + })?; + } + linker + .instance(&format!("wasi:cli/stdin@{PREVIEW2}"))? + .func_wrap("get-stdin", |mut store: Ctx<'_, P, T>, (): ()| { + Ok((store.data_mut().table().push(InputStream)?,)) + })?; + linker + .instance(&format!("wasi:cli/environment@{PREVIEW2}"))? + .func_wrap("get-environment", |_: Ctx<'_, P, T>, (): ()| { + Ok((Vec::<(String, String)>::new(),)) + })?; + linker.instance(&format!("wasi:cli/exit@{PREVIEW2}"))?.func_wrap( + "exit", + |_: Ctx<'_, P, T>, (status,): (Result<(), ()>,)| -> wasmtime::Result<()> { + let outcome = if status.is_ok() { "success" } else { "failure" }; + Err(wasmtime::Error::msg(format!("the guest exited with {outcome}"))) + }, + )?; + + linker + .instance(&format!("wasi:cli/terminal-input@{PREVIEW2}"))? + .resource("terminal-input", ResourceType::host::(), |_, _| Ok(()))?; + linker + .instance(&format!("wasi:cli/terminal-output@{PREVIEW2}"))? + .resource("terminal-output", ResourceType::host::(), |_, _| Ok(()))?; + linker + .instance(&format!("wasi:cli/terminal-stdin@{PREVIEW2}"))? + .func_wrap("get-terminal-stdin", |_: Ctx<'_, P, T>, (): ()| { + Ok((None::>,)) + })?; + for (interface, getter) in [ + ("terminal-stdout", "get-terminal-stdout"), + ("terminal-stderr", "get-terminal-stderr"), + ] { + linker + .instance(&format!("wasi:cli/{interface}@{PREVIEW2}"))? + .func_wrap(getter, |_: Ctx<'_, P, T>, (): ()| { + Ok((None::>,)) + })?; + } + Ok(()) +} + +/// Drop the guest's handle to a `R` the board created, as a resource destructor. +pub(super) fn drop_from_table(mut store: Ctx<'_, P, T>, rep: u32) -> wasmtime::Result<()> +where + R: 'static, +{ + let _dropped: R = store.data_mut().table().delete(Resource::::new_own(rep))?; + Ok(()) +} + +/// The indices of `pollables` that are ready, waiting on `platform` until at least one is. +async fn ready

(platform: &P, pollables: &[Pollable]) -> Vec +where + P: Platform, +{ + loop { + let now = platform.now(); + let mut ready = Vec::new(); + let mut earliest = u64::MAX; + for (index, pollable) in (0_u32..).zip(pollables) { + match *pollable { + Pollable::Ready => ready.push(index), + Pollable::Due(deadline) if deadline <= now => ready.push(index), + Pollable::Due(deadline) => earliest = earliest.min(deadline), + } + } + // With nothing to wait on, nothing will ever be ready, so the guest is told so rather than left waiting. + if !ready.is_empty() || earliest == u64::MAX { + return ready; + } + platform.sleep_until(earliest).await; + } +} diff --git a/services/ws-wasi-runner/src/convert.rs b/services/ws-wasi-runner/src/convert.rs new file mode 100644 index 00000000..0abb304c --- /dev/null +++ b/services/ws-wasi-runner/src/convert.rs @@ -0,0 +1,264 @@ +//! Wire messages across the WIT boundary. +//! +//! The typed `et:ws-messages/messages` values a guest sends and receives are converted to and from +//! `edge_toolkit::ws::{ClientMessage, ServerMessage}`, which every host serialises to JSON for the websocket. Guests +//! no longer hand-craft JSON; whichever transport a host uses, the conversion is the same. + +#[cfg(not(feature = "std"))] +use alloc::{ + string::{String, ToString as _}, + vec::Vec, +}; +use core::net::IpAddr; + +use chrono::{DateTime, Utc}; +use edge_toolkit::ws::{ + AgentConnectionState as EtAgentConnectionState, AgentSummary as EtAgentSummary, ClientMessage, CloudEvent, + ConnectStatus as EtConnectStatus, MessageDeliveryStatus as EtMessageDeliveryStatus, MessageScope as EtMessageScope, + ServerMessage, +}; + +use crate::bindings::et::ws_messages::messages::{ + AgentConnectionState as WitAgentConnectionState, AgentMessagePayload, AgentSummary as WitAgentSummary, + ClientMessage as WitClientMessage, ConnectAckPayload, ConnectStatus as WitConnectStatus, InvalidPayload, + IpAddress as WitIpAddress, ListAgentsResponsePayload, MessageDeliveryStatus as WitMessageDeliveryStatus, + MessageScope as WitMessageScope, MessageStatusPayload, RelayBinaryPayload, RelayTextPayload, ResponsePayload, + ServerMessage as WitServerMessage, +}; +use crate::bindings::et::ws_wasi::ws::WsError; + +/// One websocket data frame, as whichever transport a host uses carries it. +#[derive(Clone, Debug, Eq, PartialEq)] +#[non_exhaustive] +pub enum Frame { + /// A text frame: an `et-*` message's JSON, or relayed text. + Text(String), + /// A binary frame, which is always relayed. + Binary(Vec), +} + +/// The frame a guest's `client-message` goes out as. +/// +/// Relay variants travel as raw text / binary frames (no JSON envelope) -- that's what the ws-server hub forwards +/// verbatim to other agents. Typed variants serialise through standard tagged JSON. +/// +/// # Errors +/// +/// When the message's opaque JSON or event time does not parse, or the message does not serialise. +pub fn client_frame(message: WitClientMessage) -> Result { + #[expect( + clippy::wildcard_enum_match_arm, + reason = "every non-relay ClientMessage variant takes the standard tagged-JSON path" + )] + Ok(match wit_to_client_message(message)? { + ClientMessage::RelayText { content } => Frame::Text(content), + ClientMessage::RelayBinary { content } => Frame::Binary(content), + typed => Frame::Text(serde_json::to_string(&typed)?), + }) +} + +/// The message an inbound frame carries: an `et-*` message, or the frame as a relay. +/// +/// # Errors +/// +/// When a text frame claims an `et-*` type but its payload does not match it; a host drops such a frame. +pub fn server_message(frame: Frame) -> Result> { + match frame { + Frame::Text(text) => ServerMessage::from_text_frame(&text), + Frame::Binary(bytes) => Ok(ServerMessage::from_binary_frame(bytes)), + } +} + +/// Convert a guest-emitted WIT `client-message` into the canonical Rust `ClientMessage`. +/// +/// Opaque JSON fields (sent as `string` over WIT) are parsed here so the host always works with `serde_json::Value` +/// payloads. +#[expect(clippy::single_call_fn, reason = "named converter; used once by client_frame")] +fn wit_to_client_message(msg: WitClientMessage) -> Result { + let parse_value = |raw: String| -> Result { + let mut deserializer = serde_json::Deserializer::from_str(&raw); + Ok(serde_path_to_error::deserialize(&mut deserializer)?) + }; + Ok(match msg { + WitClientMessage::Connect(payload) => ClientMessage::Connect { + agent_id: payload.agent_id, + }, + WitClientMessage::Alive(payload) => ClientMessage::Alive { + timestamp: payload.timestamp, + }, + WitClientMessage::ListAgents => ClientMessage::ListAgents, + WitClientMessage::SendAgentMessage(payload) => ClientMessage::SendAgentMessage { + to_agent_id: payload.to_agent_id, + message: parse_value(payload.message)?, + }, + WitClientMessage::BroadcastMessage(payload) => ClientMessage::BroadcastMessage { + message: parse_value(payload.message)?, + }, + WitClientMessage::MessageAck(payload) => ClientMessage::MessageAck { + message_id: payload.message_id, + }, + WitClientMessage::ClientEvent(payload) => { + let event = payload.event; + let mut wire = CloudEvent::new( + event.id, + event.source, + event.type_, + DateTime::parse_from_rfc3339(&event.time)?.with_timezone(&Utc), + parse_value(event.data)?, + ); + wire.specversion = serde_json::from_value(serde_json::Value::String(event.specversion))?; + ClientMessage::ClientEvent { event: wire } + } + WitClientMessage::RelayText(payload) => ClientMessage::RelayText { + content: payload.content, + }, + WitClientMessage::RelayBinary(payload) => ClientMessage::RelayBinary { + content: payload.content, + }, + }) +} + +/// Convert a wire `ServerMessage` (inbound) into the WIT `server-message` the guest sees on `recv`. +/// +/// Re-serialises opaque JSON payloads so they cross the WIT boundary as strings. +/// +/// # Errors +/// +/// Fails when an agent message's opaque JSON payload cannot be re-serialised. +pub fn server_message_to_wit(msg: ServerMessage) -> Result { + let serialize = |value: serde_json::Value| serde_json::to_string(&value); + Ok(match msg { + ServerMessage::ConnectAck { agent_id, status } => WitServerMessage::ConnectAck(ConnectAckPayload { + agent_id, + status: et_connect_status(status), + }), + ServerMessage::ListAgentsResponse { agents } => { + WitServerMessage::ListAgentsResponse(ListAgentsResponsePayload { + agents: agents.into_iter().map(et_agent_summary).collect(), + }) + } + ServerMessage::AgentMessage { + message_id, + from_agent_id, + scope, + server_received_at, + message, + } => WitServerMessage::AgentMessage(AgentMessagePayload { + message_id, + from_agent_id, + scope: et_message_scope(scope), + server_received_at, + message: serialize(message)?, + }), + ServerMessage::MessageStatus { + message_id, + status, + detail, + } => WitServerMessage::MessageStatus(MessageStatusPayload { + message_id, + status: et_delivery_status(status), + detail, + }), + ServerMessage::Invalid { message_id, detail } => { + WitServerMessage::Invalid(InvalidPayload { detail, message_id }) + } + ServerMessage::Response { message } => WitServerMessage::Response(ResponsePayload { message }), + ServerMessage::RelayText { content } => WitServerMessage::RelayText(RelayTextPayload { content }), + ServerMessage::RelayBinary { content } => WitServerMessage::RelayBinary(RelayBinaryPayload { content }), + }) +} + +// edge-toolkit-to-WIT enum mirrors. Only the `et_*` direction is used: the host never deserialises a WIT-side client +// message back into an edge-toolkit type (that would mean reading something the *server* sent through the *guest's* +// outbound API, which never happens), so the reverse-direction helpers from the pre-split file are gone. +#[expect( + clippy::needless_pass_by_value, + clippy::single_call_fn, + reason = "WIT/edge-toolkit enum bridge; Copy enum, by value keeps call sites uniform" +)] +const fn et_connect_status(value: EtConnectStatus) -> WitConnectStatus { + match value { + EtConnectStatus::Assigned => WitConnectStatus::Assigned, + EtConnectStatus::Reconnected => WitConnectStatus::Reconnected, + } +} + +#[expect( + clippy::needless_pass_by_value, + clippy::single_call_fn, + reason = "WIT/edge-toolkit enum bridge; Copy enum, by value keeps call sites uniform" +)] +const fn et_message_scope(value: EtMessageScope) -> WitMessageScope { + match value { + EtMessageScope::Direct => WitMessageScope::Direct, + EtMessageScope::Broadcast => WitMessageScope::Broadcast, + } +} + +#[expect( + clippy::needless_pass_by_value, + clippy::single_call_fn, + reason = "WIT/edge-toolkit enum bridge; Copy enum, by value keeps call sites uniform" +)] +const fn et_delivery_status(value: EtMessageDeliveryStatus) -> WitMessageDeliveryStatus { + match value { + EtMessageDeliveryStatus::Delivered => WitMessageDeliveryStatus::Delivered, + EtMessageDeliveryStatus::Queued => WitMessageDeliveryStatus::Queued, + EtMessageDeliveryStatus::Acknowledged => WitMessageDeliveryStatus::Acknowledged, + EtMessageDeliveryStatus::Broadcast => WitMessageDeliveryStatus::Broadcast, + } +} + +#[expect( + clippy::needless_pass_by_value, + clippy::single_call_fn, + reason = "WIT/edge-toolkit enum bridge; Copy enum, by value keeps call sites uniform" +)] +const fn et_agent_connection_state(value: EtAgentConnectionState) -> WitAgentConnectionState { + match value { + EtAgentConnectionState::Connected => WitAgentConnectionState::Connected, + EtAgentConnectionState::Disconnected => WitAgentConnectionState::Disconnected, + } +} + +#[expect( + clippy::single_call_fn, + reason = "summary converter; one call site in server_message_to_wit" +)] +fn et_agent_summary(value: EtAgentSummary) -> WitAgentSummary { + WitAgentSummary { + agent_id: value.agent_id, + state: et_agent_connection_state(value.state), + last_known_ip: et_ip_address(value.last_known_ip), + } +} + +#[expect( + clippy::single_call_fn, + reason = "address converter; one call site in et_agent_summary" +)] +fn et_ip_address(value: IpAddr) -> WitIpAddress { + match value { + IpAddr::V4(addr) => WitIpAddress::Ipv4(addr.octets().into()), + IpAddr::V6(addr) => WitIpAddress::Ipv6(addr.segments().into()), + } +} + +impl From for WsError { + fn from(err: serde_json::Error) -> Self { + Self::Decode(err.to_string()) + } +} + +impl From> for WsError { + fn from(err: serde_path_to_error::Error) -> Self { + Self::Decode(err.to_string()) + } +} + +/// A guest event `time` that is not an RFC 3339 timestamp cannot be encoded onto the wire. +impl From for WsError { + fn from(err: chrono::ParseError) -> Self { + Self::Decode(err.to_string()) + } +} diff --git a/services/ws-wasi-runner/src/engine.rs b/services/ws-wasi-runner/src/engine.rs new file mode 100644 index 00000000..4d4cf026 --- /dev/null +++ b/services/ws-wasi-runner/src/engine.rs @@ -0,0 +1,76 @@ +//! The engine a board runs a component with, and compiling a component for it ahead of time. +//! +//! A board has no compiler and no virtual memory, so its engine runs Pulley bytecode a host compiled for it, keeps +//! each linear memory to what it holds, and checks bounds and traps explicitly rather than through signals. The host +//! has to compile with the same settings the board loads with, so both come from [`board_config`]. + +#[cfg(all(feature = "precompile", not(feature = "std")))] +use alloc::vec::Vec; + +use wasmtime::{Config, Engine}; + +/// The Pulley target a board with this build's pointer width runs. +pub const NATIVE_PULLEY: &str = if cfg!(target_pointer_width = "64") { + "pulley64" +} else { + "pulley32" +}; +/// The stack each guest call runs on, the most a board gives one. +/// +/// The interpreter and the host functions a guest calls run on it too, and unoptimised their frames are several +/// times larger, so a debug build gives it four times the room. +const ASYNC_STACK_SIZE: usize = if cfg!(debug_assertions) { + // 64 KiB overflowed on Windows, on `override (msvc)` at commit + // https://github.com/edge-toolkit/core/commit/0a3156acdc5f4405cb318451a58ace3c53922d5c + // (`https://github.com/edge-toolkit/core/actions/runs/41308264184/job/124853064168`): + // + // thread 'data1_round_trips_a_value_through_the_boards_memory' (876) has overflowed its stack + 256 +} else { + 64 +} * 1024; +/// The guest's share of [`ASYNC_STACK_SIZE`]; the rest is the host functions' it calls. +const MAX_WASM_STACK: usize = 32 * 1024; + +/// The settings a board's engine runs with, compiling for `target`. +/// +/// # Errors +/// +/// When `target` names no target this build of wasmtime knows. +pub fn board_config(target: &str) -> wasmtime::Result { + let mut config = Config::new(); + #[expect( + unused_results, + reason = "each setter returns &mut Config for chaining; the mutation is the intent" + )] + { + config.target(target)?; + config.wasm_component_model_async(true); + config.signals_based_traps(false); + config.memory_reservation(0); + config.memory_reservation_for_growth(0); + config.memory_guard_size(0); + config.async_stack_size(ASYNC_STACK_SIZE); + config.max_wasm_stack(MAX_WASM_STACK); + } + Ok(config) +} + +/// The engine a board runs components with, which loads only what [`precompile`] compiled for this build's target. +/// +/// # Errors +/// +/// When wasmtime rejects the settings, as it would one that this build cannot run. +pub fn board_engine() -> wasmtime::Result { + Engine::new(&board_config(NATIVE_PULLEY)?) +} + +/// `component`, compiled for a board running `target`, as that board's [`board_engine`] loads it. +/// +/// # Errors +/// +/// When `target` is unknown, or `component` is not a valid component. +#[cfg(feature = "precompile")] +pub fn precompile(target: &str, component: &[u8]) -> wasmtime::Result> { + Engine::new(&board_config(target)?)?.precompile_component(component) +} diff --git a/services/ws-wasi-runner/src/host/error.rs b/services/ws-wasi-runner/src/host/error.rs index 0e8fb221..cca28835 100644 --- a/services/ws-wasi-runner/src/host/error.rs +++ b/services/ws-wasi-runner/src/host/error.rs @@ -1,18 +1,12 @@ -//! Error conversions for the runner's host impls. +//! Error conversions for the runner's std host impls: from its HTTP client and its websocket. use crate::bindings::et::ws_wasi::ws::WsError; use crate::bindings::wasi::keyvalue::store::Error as KvError; /// Conversions into `wasi:keyvalue/store`'s `error.other(message)` variant. /// -/// One impl per concrete source type: the resource table, the REST client's byte stream, and its -/// request errors are the only things the `keyvalue` host impls can fail on. -impl From for KvError { - fn from(err: wasmtime::component::ResourceTableError) -> Self { - Self::Other(err.to_string()) - } -} - +/// One impl per concrete source type: the REST client's byte stream and its request errors are what the +/// `keyvalue` host impls can fail on besides the resource table. impl From for KvError { fn from(err: reqwest::Error) -> Self { Self::Other(err.to_string()) @@ -25,13 +19,6 @@ impl From> for KvError { } } -/// Build a `wasi:keyvalue/store.error.other(" not implemented")`. -/// That is the closest thing the WIT-spec enum has to a `NotImplemented` variant. -#[must_use] -pub fn kv_not_implemented(operation: &str) -> KvError { - KvError::Other(format!("{operation} not implemented")) -} - /// Maps a `tokio_tungstenite::tungstenite::Error` into a typed `WsError`. /// /// Specialised to `tungstenite::Error` so the variant carries through: @@ -55,25 +42,6 @@ impl From for WsError { } } -impl From for WsError { - fn from(err: serde_json::Error) -> Self { - Self::Decode(err.to_string()) - } -} - -impl From> for WsError { - fn from(err: serde_path_to_error::Error) -> Self { - Self::Decode(err.to_string()) - } -} - -/// A guest event `time` that is not an RFC 3339 timestamp cannot be encoded onto the wire. -impl From for WsError { - fn from(err: chrono::ParseError) -> Self { - Self::Decode(err.to_string()) - } -} - impl From for WsError { fn from(err: et_ws_runner_common::ConnectError) -> Self { Self::Transport(err.to_string()) diff --git a/services/ws-wasi-runner/src/host/mod.rs b/services/ws-wasi-runner/src/host/mod.rs index 40e746a3..6dc897cb 100644 --- a/services/ws-wasi-runner/src/host/mod.rs +++ b/services/ws-wasi-runner/src/host/mod.rs @@ -7,12 +7,13 @@ mod error; mod log; +mod run; mod state; pub mod wasi_keyvalue; #[cfg(feature = "nn")] pub mod wasi_nn; pub mod ws; -pub use self::error::kv_not_implemented; +pub use self::run::run_module; pub use self::state::HostState; pub use self::ws::WsBackend; diff --git a/services/ws-wasi-runner/src/host/run.rs b/services/ws-wasi-runner/src/host/run.rs new file mode 100644 index 00000000..4c0c6f21 --- /dev/null +++ b/services/ws-wasi-runner/src/host/run.rs @@ -0,0 +1,89 @@ +//! Fetch a module's WASI component from the hub, compile it, and run it with WASI Preview 2 and the `et:ws-wasi` +//! world hosted. + +use et_ws_runner_common::{derive_http_base, fetch_main_field}; +use tracing::{Instrument as _, info, info_span}; +use wasmtime::component::{Component, HasSelf, Linker}; +use wasmtime::{Config, Engine, Store}; + +use crate::{HostState, RunnerError, bindings}; + +/// Download, link, and run the WASI component for `module_name`. +/// +/// Returns when the guest's exported `entry.run` finishes (either by +/// returning `ok` or trapping). Guest `err` returns are surfaced as +/// `RunnerError::Guest`. +/// +/// The whole call is wrapped in a `run_module` span -- every outgoing +/// request inherits its trace context, and ws-server's request span ends +/// up as a child of it. +pub async fn run_module( + module_name: &str, + ws_url: &str, + connect_ack_timeout: Option, + coverage: bool, +) -> Result<(), RunnerError> { + let span = info_span!("run_module", module = module_name); + run_module_inner(module_name, ws_url, connect_ack_timeout, coverage) + .instrument(span) + .await +} + +#[expect( + clippy::cognitive_complexity, + clippy::single_call_fn, + reason = "span-instrumented body of run_module, split to scope the span; its score is mostly info! expansion" +)] +async fn run_module_inner( + module_name: &str, + ws_url: &str, + connect_ack_timeout: Option, + coverage: bool, +) -> Result<(), RunnerError> { + let http_base = derive_http_base(ws_url)?; + + let rest = et_rest_client::Client::new(&http_base); + let main = fetch_main_field(&rest, module_name).await?; + info!(module = module_name, %main, "fetching WASI component"); + let wasm_bytes = et_ws_runner_common::fetch_module_file(&rest, module_name, &main) + .instrument(info_span!("fetch_component", module = module_name, file = %main)) + .await?; + + let mut config = Config::new(); + #[expect( + unused_results, + reason = "wasmtime::Config::wasm_component_model returns &mut Self for builder chaining; mutation is the intent" + )] + { + config.wasm_component_model(true); + // wasi:webgpu@0.3.0-rc.2 declares `request-adapter`, `request-device` and `map-async` as + // `async func`, which is component-model-async (WASI Preview 3) ABI. Without this the guest's + // webgpu imports fail to instantiate. + config.wasm_component_model_async(true); + } + let engine = Engine::new(&config)?; + + let component = Component::from_binary(&engine, &wasm_bytes)?; + + let mut linker: Linker = Linker::new(&engine); + wasmtime_wasi::p2::add_to_linker_async(&mut linker)?; + bindings::Runner::add_to_linker::>(&mut linker, |state| state)?; + #[cfg(feature = "nn")] + wasmtime_wasi_nn::wit::add_to_linker(&mut linker, super::wasi_nn::view)?; + #[cfg(feature = "webgpu")] + wasi_webgpu_wasmtime::add_to_linker(&mut linker)?; + + let host_state = HostState::new(&http_base, ws_url.to_string(), connect_ack_timeout, coverage); + let mut store = Store::new(&engine, host_state); + + let module = bindings::Runner::instantiate_async(&mut store, &component, &linker).await?; + + // `entry.run` is an async export, so it is driven through the concurrent API rather than called + // directly against the store: the guest may await a host import (wasi-webgpu's `request-adapter`, + // `request-device`, `map-async`) mid-call, and only an `Accessor` can hand the store back to the + // host while the guest task is suspended. + store + .run_concurrent(async |accessor| module.et_ws_wasi_entry().call_run(accessor).await) + .await???; + Ok(()) +} diff --git a/services/ws-wasi-runner/src/host/wasi_keyvalue.rs b/services/ws-wasi-runner/src/host/wasi_keyvalue.rs index 261e0196..d96951cf 100644 --- a/services/ws-wasi-runner/src/host/wasi_keyvalue.rs +++ b/services/ws-wasi-runner/src/host/wasi_keyvalue.rs @@ -1,72 +1,12 @@ //! Implements `wasi:keyvalue/store` against the ws-server's storage and modules services via the typed -//! `et-rest-client`. The bucket identifier names a namespace: -//! -//! * `` -> per-agent storage bucket. Reads work for any agent's bucket (server static-serves everything -//! under `/storage/`); writes only succeed when the runner's own agent owns the bucket (server enforces `agent_id` -//! is registered). -//! * `modules/` -> module asset bucket. Used by guests to fetch their own static assets bundled in -//! `pkg/`. Writes return `access-denied` since et-modules-service serves files static. +//! `et-rest-client`, for the buckets [`crate::keyvalue`] names. use futures_util::StreamExt as _; use wasmtime::component::Resource; use crate::HostState; use crate::bindings::wasi::keyvalue::store::{Error, Host, HostBucket, KeyResponse}; -use crate::host::kv_not_implemented; - -/// Bucket-kind discriminator. -/// -/// The wire prefix on the ws-server is implied by the variant; the typed REST client picks the right operation. -#[non_exhaustive] -pub enum Bucket { - /// `/storage/{agent_id}/` -- writable, owned by the named agent. - Storage { agent_id: String }, - /// `/modules/{module_name}/` -- read-only static module assets. - Modules { module_name: String }, -} - -/// True when `candidate` names one module, rather than addressing a path through one. -/// -/// A module is served under the name it publishes under, so a scoped name carries exactly one `/`, directly -/// after a leading `@`. Any other separator is a path, which a bucket identifier may not express -- the key -/// passed to `get` is what addresses a file inside the bucket. -#[expect( - clippy::single_call_fn, - reason = "named predicate; used once by bucket_from_identifier and kept separate so the shape is readable" -)] -fn names_one_module(candidate: &str) -> bool { - let Some(scoped) = candidate.strip_prefix('@') else { - return !candidate.is_empty() && !candidate.contains('/'); - }; - match scoped.split_once('/') { - Some((scope, name)) => !scope.is_empty() && !name.is_empty() && !name.contains('/'), - None => false, - } -} - -/// Map a `store.open` identifier to a bucket variant. -#[expect( - clippy::single_call_fn, - reason = "named identifier parser; used once by ::open" -)] -fn bucket_from_identifier(identifier: &str) -> Result { - if let Some(module_name) = identifier.strip_prefix("modules/") { - if !names_one_module(module_name) { - return Err(Error::Other(format!( - "invalid module bucket identifier: {identifier:?}" - ))); - } - return Ok(Bucket::Modules { - module_name: module_name.to_string(), - }); - } - if identifier.is_empty() || identifier.contains('/') { - return Err(Error::Other(format!("invalid bucket identifier: {identifier:?}"))); - } - Ok(Bucket::Storage { - agent_id: identifier.to_string(), - }) -} +use crate::keyvalue::{Bucket, bucket_from_identifier, kv_not_implemented}; /// Drain a progenitor `ByteStream` into a `Vec`. /// diff --git a/services/ws-wasi-runner/src/host/ws.rs b/services/ws-wasi-runner/src/host/ws.rs index aecfc9ed..053b29c7 100644 --- a/services/ws-wasi-runner/src/host/ws.rs +++ b/services/ws-wasi-runner/src/host/ws.rs @@ -1,22 +1,13 @@ //! Implements `et:ws-wasi/ws` using `tokio-tungstenite`. //! //! On `connect`, we open a websocket, send `ClientMessage::Connect { agent_id: None }`, and spawn a task that pumps -//! inbound text messages into a channel. Inbound `connect_ack` messages capture our assigned `agent_id`. -//! -//! Wire messages cross the WIT boundary as typed `et:ws-messages/messages.ws-message` values. The host converts them -//! to/from `edge_toolkit::ws::{ClientMessage, ServerMessage}` and serialises to JSON for the actual websocket frame. -//! Guests no longer hand-craft JSON. +//! inbound text messages into a channel. Inbound `connect_ack` messages capture our assigned `agent_id`. Messages +//! cross the WIT boundary as [`crate::convert`] converts them. -use std::net::IpAddr; use std::sync::Arc; use std::time::Duration; -use chrono::{DateTime, Utc}; -use edge_toolkit::ws::{ - AgentConnectionState as EtAgentConnectionState, AgentSummary as EtAgentSummary, ClientMessage, CloudEvent, - ConnectStatus as EtConnectStatus, MessageDeliveryStatus as EtMessageDeliveryStatus, MessageScope as EtMessageScope, - ServerMessage, -}; +use edge_toolkit::ws::ServerMessage; use futures_util::SinkExt as _; use futures_util::stream::{SplitSink, StreamExt as _}; use tokio::net::TcpStream; @@ -27,13 +18,10 @@ use tracing::warn; use crate::HostState; use crate::bindings::et::ws_messages::messages::{ - AgentConnectionState as WitAgentConnectionState, AgentSummary as WitAgentSummary, - ClientMessage as WitClientMessage, ConnectAckPayload, ConnectStatus as WitConnectStatus, InvalidPayload, - IpAddress as WitIpAddress, ListAgentsResponsePayload, MessageDeliveryStatus as WitMessageDeliveryStatus, - MessageScope as WitMessageScope, MessageStatusPayload, RelayBinaryPayload, RelayTextPayload, ResponsePayload, - ServerMessage as WitServerMessage, + ClientMessage as WitClientMessage, ServerMessage as WitServerMessage, }; use crate::bindings::et::ws_wasi::ws::{Host, State, WsError}; +use crate::convert::{Frame, client_frame, server_message, server_message_to_wit}; // The `et:ws-messages/messages` interface only declares types -- no functions. wasmtime-bindgen still requires a `Host` // impl so the linker has somewhere to anchor the interface. The trait body is empty. @@ -82,9 +70,9 @@ impl WsBackend { let agent_id = Arc::new(Mutex::new(Some(assigned_id))); let connection_state = Arc::new(Mutex::new(State::Connected)); - // Reader pump: convert every subsequent Text/Binary data frame into a `ServerMessage` via - // `ServerMessage::from_*_frame` (foreign frames land in `RelayText`/`RelayBinary`) and forward it to the guest - // inbox; drop control frames and et-prefixed-but-malformed text with a warn. + // Reader pump: convert every subsequent Text/Binary data frame into a `ServerMessage` (foreign frames land in + // `RelayText`/`RelayBinary`) and forward it to the guest inbox; drop control frames and + // et-prefixed-but-malformed text with a warn. let state_clone = Arc::clone(&connection_state); let reader = tokio::spawn(async move { while let Some(msg) = stream.next().await { @@ -95,17 +83,18 @@ impl WsBackend { clippy::wildcard_enum_match_arm, reason = "control frames (Ping/Pong/Close/Frame) and any future variants are dropped" )] - let parsed = match msg { - tungstenite::Message::Text(text) => match ServerMessage::from_text_frame(text.as_str()) { - Ok(msg) => msg, - Err(err) => { - warn!(error = %err, "dropping et-* frame with decode error"); - continue; - } - }, - tungstenite::Message::Binary(bytes) => ServerMessage::from_binary_frame(bytes.to_vec()), + let frame = match msg { + tungstenite::Message::Text(text) => Frame::Text(text.as_str().to_owned()), + tungstenite::Message::Binary(bytes) => Frame::Binary(bytes.to_vec()), _ => continue, }; + let parsed = match server_message(frame) { + Ok(msg) => msg, + Err(err) => { + warn!(error = %err, "dropping et-* frame with decode error"); + continue; + } + }; if tx.send(parsed).is_err() { break; } @@ -204,20 +193,9 @@ impl Host for HostState { } async fn send(&mut self, message: WitClientMessage) -> Result<(), WsError> { - let et_message = wit_to_client_message(message)?; - // Relay variants travel as raw text / binary frames (no JSON envelope) -- that's what the ws-server hub - // forwards verbatim to other agents. Typed variants serialise through standard tagged JSON. - #[expect( - clippy::wildcard_enum_match_arm, - reason = "every non-relay ClientMessage variant takes the standard tagged-JSON path" - )] - let outgoing = match et_message { - ClientMessage::RelayText { content } => tungstenite::Message::text(content), - ClientMessage::RelayBinary { content } => tungstenite::Message::binary(content), - typed => { - let payload = serde_json::to_string(&typed)?; - tungstenite::Message::text(payload) - } + let outgoing = match client_frame(message)? { + Frame::Text(text) => tungstenite::Message::text(text), + Frame::Binary(bytes) => tungstenite::Message::binary(bytes), }; // Clone the sink Arc and release the outer lock before awaiting the inner send -- keeping `self.ws` locked // across the await would block every other ws-method call. @@ -259,181 +237,3 @@ impl Host for HostState { *slot = None; } } - -/// Convert a guest-emitted WIT `client-message` into the canonical Rust `ClientMessage`. -/// -/// Opaque JSON fields (sent as `string` over WIT) are parsed here so the host always works with `serde_json::Value` -/// payloads. -#[expect( - clippy::single_call_fn, - reason = "named converter; used once by ::send" -)] -fn wit_to_client_message(msg: WitClientMessage) -> Result { - let parse_value = |raw: String| -> Result { - let mut deserializer = serde_json::Deserializer::from_str(&raw); - Ok(serde_path_to_error::deserialize(&mut deserializer)?) - }; - Ok(match msg { - WitClientMessage::Connect(payload) => ClientMessage::Connect { - agent_id: payload.agent_id, - }, - WitClientMessage::Alive(payload) => ClientMessage::Alive { - timestamp: payload.timestamp, - }, - WitClientMessage::ListAgents => ClientMessage::ListAgents, - WitClientMessage::SendAgentMessage(payload) => ClientMessage::SendAgentMessage { - to_agent_id: payload.to_agent_id, - message: parse_value(payload.message)?, - }, - WitClientMessage::BroadcastMessage(payload) => ClientMessage::BroadcastMessage { - message: parse_value(payload.message)?, - }, - WitClientMessage::MessageAck(payload) => ClientMessage::MessageAck { - message_id: payload.message_id, - }, - WitClientMessage::ClientEvent(payload) => { - let event = payload.event; - let mut wire = CloudEvent::new( - event.id, - event.source, - event.type_, - DateTime::parse_from_rfc3339(&event.time)?.with_timezone(&Utc), - parse_value(event.data)?, - ); - wire.specversion = serde_json::from_value(serde_json::Value::String(event.specversion))?; - ClientMessage::ClientEvent { event: wire } - } - WitClientMessage::RelayText(payload) => ClientMessage::RelayText { - content: payload.content, - }, - WitClientMessage::RelayBinary(payload) => ClientMessage::RelayBinary { - content: payload.content, - }, - }) -} - -/// Convert a wire `ServerMessage` (inbound) into the WIT `server-message` the guest sees on `recv`. -/// -/// Re-serialises opaque JSON payloads so they cross the WIT boundary as strings. -/// -/// # Errors -/// -/// Fails when an agent message's opaque JSON payload cannot be re-serialised. -pub fn server_message_to_wit(msg: ServerMessage) -> Result { - let serialize = |value: serde_json::Value| serde_json::to_string(&value); - Ok(match msg { - ServerMessage::ConnectAck { agent_id, status } => WitServerMessage::ConnectAck(ConnectAckPayload { - agent_id, - status: et_connect_status(status), - }), - ServerMessage::ListAgentsResponse { agents } => { - WitServerMessage::ListAgentsResponse(ListAgentsResponsePayload { - agents: agents.into_iter().map(et_agent_summary).collect(), - }) - } - ServerMessage::AgentMessage { - message_id, - from_agent_id, - scope, - server_received_at, - message, - } => WitServerMessage::AgentMessage(crate::bindings::et::ws_messages::messages::AgentMessagePayload { - message_id, - from_agent_id, - scope: et_message_scope(scope), - server_received_at, - message: serialize(message)?, - }), - ServerMessage::MessageStatus { - message_id, - status, - detail, - } => WitServerMessage::MessageStatus(MessageStatusPayload { - message_id, - status: et_delivery_status(status), - detail, - }), - ServerMessage::Invalid { message_id, detail } => { - WitServerMessage::Invalid(InvalidPayload { detail, message_id }) - } - ServerMessage::Response { message } => WitServerMessage::Response(ResponsePayload { message }), - ServerMessage::RelayText { content } => WitServerMessage::RelayText(RelayTextPayload { content }), - ServerMessage::RelayBinary { content } => WitServerMessage::RelayBinary(RelayBinaryPayload { content }), - }) -} - -// edge-toolkit-to-WIT enum mirrors. Only the `et_*` direction is used: the host never deserialises a WIT-side client -// message back into an edge-toolkit type (that would mean reading something the *server* sent through the *guest's* -// outbound API, which never happens), so the reverse-direction helpers from the pre-split file are gone. -#[expect( - clippy::needless_pass_by_value, - clippy::single_call_fn, - reason = "WIT/edge-toolkit enum bridge; Copy enum, by value keeps call sites uniform" -)] -const fn et_connect_status(value: EtConnectStatus) -> WitConnectStatus { - match value { - EtConnectStatus::Assigned => WitConnectStatus::Assigned, - EtConnectStatus::Reconnected => WitConnectStatus::Reconnected, - } -} - -#[expect( - clippy::needless_pass_by_value, - clippy::single_call_fn, - reason = "WIT/edge-toolkit enum bridge; Copy enum, by value keeps call sites uniform" -)] -const fn et_message_scope(value: EtMessageScope) -> WitMessageScope { - match value { - EtMessageScope::Direct => WitMessageScope::Direct, - EtMessageScope::Broadcast => WitMessageScope::Broadcast, - } -} - -#[expect( - clippy::needless_pass_by_value, - clippy::single_call_fn, - reason = "WIT/edge-toolkit enum bridge; Copy enum, by value keeps call sites uniform" -)] -const fn et_delivery_status(value: EtMessageDeliveryStatus) -> WitMessageDeliveryStatus { - match value { - EtMessageDeliveryStatus::Delivered => WitMessageDeliveryStatus::Delivered, - EtMessageDeliveryStatus::Queued => WitMessageDeliveryStatus::Queued, - EtMessageDeliveryStatus::Acknowledged => WitMessageDeliveryStatus::Acknowledged, - EtMessageDeliveryStatus::Broadcast => WitMessageDeliveryStatus::Broadcast, - } -} - -#[expect( - clippy::needless_pass_by_value, - clippy::single_call_fn, - reason = "WIT/edge-toolkit enum bridge; Copy enum, by value keeps call sites uniform" -)] -const fn et_agent_connection_state(value: EtAgentConnectionState) -> WitAgentConnectionState { - match value { - EtAgentConnectionState::Connected => WitAgentConnectionState::Connected, - EtAgentConnectionState::Disconnected => WitAgentConnectionState::Disconnected, - } -} - -#[expect( - clippy::single_call_fn, - reason = "summary converter; one call site in server_message_to_wit" -)] -fn et_agent_summary(value: EtAgentSummary) -> WitAgentSummary { - WitAgentSummary { - agent_id: value.agent_id, - state: et_agent_connection_state(value.state), - last_known_ip: et_ip_address(value.last_known_ip), - } -} - -#[expect( - clippy::single_call_fn, - reason = "address converter; one call site in et_agent_summary" -)] -fn et_ip_address(value: IpAddr) -> WitIpAddress { - match value { - IpAddr::V4(addr) => WitIpAddress::Ipv4(addr.octets().into()), - IpAddr::V6(addr) => WitIpAddress::Ipv6(addr.segments().into()), - } -} diff --git a/services/ws-wasi-runner/src/keyvalue.rs b/services/ws-wasi-runner/src/keyvalue.rs new file mode 100644 index 00000000..1da2cc69 --- /dev/null +++ b/services/ws-wasi-runner/src/keyvalue.rs @@ -0,0 +1,86 @@ +//! `wasi:keyvalue/store` buckets as every host names them: what a `store.open` identifier opens, and the errors a +//! bucket call can fail with. +//! +//! The bucket identifier names a namespace: +//! +//! * `` -> per-agent storage bucket. Reads work for any agent's bucket (server static-serves everything +//! under `/storage/`); writes only succeed when the runner's own agent owns the bucket (server enforces `agent_id` +//! is registered). +//! * `modules/` -> module asset bucket. Used by guests to fetch their own static assets bundled in +//! `pkg/`. Writes return `access-denied` since et-modules-service serves files static. + +#[cfg(not(feature = "std"))] +use alloc::{ + format, + string::{String, ToString as _}, +}; + +use crate::bindings::wasi::keyvalue::store::Error; + +/// Bucket-kind discriminator. +/// +/// The wire prefix on the ws-server is implied by the variant; the typed REST client picks the right operation. +#[non_exhaustive] +pub enum Bucket { + /// `/storage/{agent_id}/` -- writable, owned by the named agent. + Storage { agent_id: String }, + /// `/modules/{module_name}/` -- read-only static module assets. + Modules { module_name: String }, +} + +/// True when `candidate` names one module, rather than addressing a path through one. +/// +/// A module is served under the name it publishes under, so a scoped name carries exactly one `/`, directly +/// after a leading `@`. Any other separator is a path, which a bucket identifier may not express -- the key +/// passed to `get` is what addresses a file inside the bucket. +#[expect( + clippy::single_call_fn, + reason = "named predicate; used once by bucket_from_identifier and kept separate so the shape is readable" +)] +fn names_one_module(candidate: &str) -> bool { + let Some(scoped) = candidate.strip_prefix('@') else { + return !candidate.is_empty() && !candidate.contains('/'); + }; + match scoped.split_once('/') { + Some((scope, name)) => !scope.is_empty() && !name.is_empty() && !name.contains('/'), + None => false, + } +} + +/// Map a `store.open` identifier to a bucket variant. +/// +/// # Errors +/// +/// When `identifier` is empty, or is a path rather than one agent's id or one module's name. +pub fn bucket_from_identifier(identifier: &str) -> Result { + if let Some(module_name) = identifier.strip_prefix("modules/") { + if !names_one_module(module_name) { + return Err(Error::Other(format!( + "invalid module bucket identifier: {identifier:?}" + ))); + } + return Ok(Bucket::Modules { + module_name: module_name.to_string(), + }); + } + if identifier.is_empty() || identifier.contains('/') { + return Err(Error::Other(format!("invalid bucket identifier: {identifier:?}"))); + } + Ok(Bucket::Storage { + agent_id: identifier.to_string(), + }) +} + +/// Build a `wasi:keyvalue/store.error.other(" not implemented")`. +/// That is the closest thing the WIT-spec enum has to a `NotImplemented` variant. +#[must_use] +pub fn kv_not_implemented(operation: &str) -> Error { + Error::Other(format!("{operation} not implemented")) +} + +/// A bucket handle the guest passed that its store no longer holds. +impl From for Error { + fn from(err: wasmtime::component::ResourceTableError) -> Self { + Self::Other(err.to_string()) + } +} diff --git a/services/ws-wasi-runner/src/lib.rs b/services/ws-wasi-runner/src/lib.rs index 797eb969..90fc47b2 100644 --- a/services/ws-wasi-runner/src/lib.rs +++ b/services/ws-wasi-runner/src/lib.rs @@ -1,96 +1,23 @@ #![cfg_attr(doc, doc = include_str!("../README.md"))] #![cfg_attr(feature = "docs", doc = "## Feature flags")] #![cfg_attr(feature = "docs", doc = document_features::document_features!())] +#![cfg_attr(not(feature = "std"), no_std)] -use et_ws_runner_common::{derive_http_base, fetch_main_field}; -use tracing::{Instrument as _, info, info_span}; -use wasmtime::component::{Component, HasSelf, Linker}; -use wasmtime::{Config, Engine, Store}; +extern crate alloc; pub mod bindings; +pub mod board; +#[cfg(feature = "std")] pub mod config; +pub mod convert; +pub mod engine; +#[cfg(feature = "std")] pub mod error; +#[cfg(feature = "std")] pub mod host; +pub mod keyvalue; +#[cfg(feature = "std")] pub use self::error::RunnerError; -pub use self::host::HostState; - -/// Download, link, and run the WASI component for `module_name`. -/// -/// Returns when the guest's exported `entry.run` finishes (either by -/// returning `ok` or trapping). Guest `err` returns are surfaced as -/// `RunnerError::Guest`. -/// -/// The whole call is wrapped in a `run_module` span -- every outgoing -/// request inherits its trace context, and ws-server's request span ends -/// up as a child of it. -pub async fn run_module( - module_name: &str, - ws_url: &str, - connect_ack_timeout: Option, - coverage: bool, -) -> Result<(), RunnerError> { - let span = info_span!("run_module", module = module_name); - run_module_inner(module_name, ws_url, connect_ack_timeout, coverage) - .instrument(span) - .await -} - -#[expect( - clippy::cognitive_complexity, - clippy::single_call_fn, - reason = "span-instrumented body of run_module, split to scope the span; its score is mostly info! expansion" -)] -async fn run_module_inner( - module_name: &str, - ws_url: &str, - connect_ack_timeout: Option, - coverage: bool, -) -> Result<(), RunnerError> { - let http_base = derive_http_base(ws_url)?; - - let rest = et_rest_client::Client::new(&http_base); - let main = fetch_main_field(&rest, module_name).await?; - info!(module = module_name, %main, "fetching WASI component"); - let wasm_bytes = et_ws_runner_common::fetch_module_file(&rest, module_name, &main) - .instrument(info_span!("fetch_component", module = module_name, file = %main)) - .await?; - - let mut config = Config::new(); - #[expect( - unused_results, - reason = "wasmtime::Config::wasm_component_model returns &mut Self for builder chaining; mutation is the intent" - )] - { - config.wasm_component_model(true); - // wasi:webgpu@0.3.0-rc.2 declares `request-adapter`, `request-device` and `map-async` as - // `async func`, which is component-model-async (WASI Preview 3) ABI. Without this the guest's - // webgpu imports fail to instantiate. - config.wasm_component_model_async(true); - } - let engine = Engine::new(&config)?; - - let component = Component::from_binary(&engine, &wasm_bytes)?; - - let mut linker: Linker = Linker::new(&engine); - wasmtime_wasi::p2::add_to_linker_async(&mut linker)?; - bindings::Runner::add_to_linker::>(&mut linker, |state| state)?; - #[cfg(feature = "nn")] - wasmtime_wasi_nn::wit::add_to_linker(&mut linker, host::wasi_nn::view)?; - #[cfg(feature = "webgpu")] - wasi_webgpu_wasmtime::add_to_linker(&mut linker)?; - - let host_state = HostState::new(&http_base, ws_url.to_string(), connect_ack_timeout, coverage); - let mut store = Store::new(&engine, host_state); - - let module = bindings::Runner::instantiate_async(&mut store, &component, &linker).await?; - - // `entry.run` is an async export, so it is driven through the concurrent API rather than called - // directly against the store: the guest may await a host import (wasi-webgpu's `request-adapter`, - // `request-device`, `map-async`) mid-call, and only an `Accessor` can hand the store back to the - // host while the guest task is suspended. - store - .run_concurrent(async |accessor| module.et_ws_wasi_entry().call_run(accessor).await) - .await???; - Ok(()) -} +#[cfg(feature = "std")] +pub use self::host::{HostState, run_module}; diff --git a/services/ws-wasi-runner/tests/board.rs b/services/ws-wasi-runner/tests/board.rs new file mode 100644 index 00000000..77ae8c8b --- /dev/null +++ b/services/ws-wasi-runner/tests/board.rs @@ -0,0 +1,336 @@ +//! Runs guests through the board's runner on this host: compiled ahead of time for the board engine this build has, +//! with a hub played in memory, a clock that jumps to each deadline a guest sleeps until, and a log kept in a vector. +#![cfg(test)] + +use std::collections::VecDeque; +use std::net::{IpAddr, Ipv4Addr}; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, Mutex}; + +use edge_toolkit::config::get_project_root; +use edge_toolkit::ws::{ + AgentConnectionState, AgentSummary, ClientMessage, ConnectStatus, MessageDeliveryStatus, MessageScope, + ServerMessage, +}; +use et_ws_wasi_runner::bindings::et::ws_wasi::ws::WsError; +use et_ws_wasi_runner::bindings::exports::et::ws_wasi::entry::EntryError; +use et_ws_wasi_runner::bindings::wasi::logging::logging::Level; +use et_ws_wasi_runner::board::{BoardError, Console, Platform, Transport, run}; +use et_ws_wasi_runner::convert::Frame; +use et_ws_wasi_runner::engine::{NATIVE_PULLEY, board_engine, precompile}; +use rstest::rstest; + +/// The id the hub assigns the guest. +const AGENT_ID: &str = "board-agent"; + +/// A board's clock and log, played by this host. +/// +/// The clock only moves when the guest sleeps, and the log keeps every line logged or written, in order. +struct Host { + clock: AtomicU64, + lines: Arc>>, +} + +impl Platform for Host { + fn now(&self) -> u64 { + self.clock.load(Ordering::SeqCst) + } + + async fn sleep_until(&self, deadline: u64) { + let _before = self.clock.fetch_max(deadline, Ordering::SeqCst); + } + + fn log(&self, _level: Level, _context: &str, message: &str) { + self.lines.lock().unwrap().push(message.to_owned()); + } + + fn write(&self, _console: Console, bytes: &[u8]) { + self.lines + .lock() + .unwrap() + .push(String::from_utf8_lossy(bytes).into_owned()); + } +} + +/// Where the hub's websocket fails, if it does. +#[derive(Clone, Copy, Default, Eq, PartialEq)] +enum Fault { + /// Nowhere. + #[default] + None, + /// It cannot be opened. + Refused, + /// Every frame sent fails, the connect's included. + SendFails, + /// Every frame sent after the connect fails. + SendFailsAfterConnect, + /// Every read fails. + Unreadable, + /// Every read fails once the frames the hub queued have all been read. + RecvFailsWhenIdle, + /// The hub never acknowledges the connect. + NoAck, +} + +/// A hub that answers a connect with an id after the frames it was given. +/// +/// It records every frame it is sent, and whether it was closed. +#[derive(Default)] +struct Hub { + sent: Arc>>, + closed: Arc>, + replies: VecDeque, + fault: Fault, +} + +impl Hub { + /// A hub that fails at `fault`, and sends `frames` ahead of its connect ack. + fn new(fault: Fault, frames: impl IntoIterator) -> Self { + Self { + replies: frames.into_iter().collect(), + fault, + ..Self::default() + } + } +} + +#[expect( + clippy::unused_async_trait_impl, + reason = "the hub answers at once from memory, where a board's transport awaits its socket" +)] +impl Transport for Hub { + type Error = &'static str; + + async fn connect(&mut self) -> Result<(), &'static str> { + if self.fault == Fault::Refused { + return Err("refused"); + } + Ok(()) + } + + async fn send(&mut self, frame: Frame) -> Result<(), &'static str> { + let first = self.sent.lock().unwrap().is_empty(); + if self.fault == Fault::SendFails || (self.fault == Fault::SendFailsAfterConnect && !first) { + return Err("broken pipe"); + } + if let Frame::Text(text) = &frame { + let reply = match serde_json::from_str(text) { + Ok(ClientMessage::Connect { .. }) if self.fault != Fault::NoAck => Some(ServerMessage::ConnectAck { + agent_id: AGENT_ID.to_owned(), + status: ConnectStatus::Assigned, + }), + Ok(ClientMessage::ListAgents) => Some(ServerMessage::ListAgentsResponse { + agents: vec![AgentSummary::new( + AGENT_ID.to_owned(), + AgentConnectionState::Connected, + IpAddr::V4(Ipv4Addr::LOCALHOST), + )], + }), + _ => None, + }; + if let Some(reply) = reply { + self.replies.push_back(text_frame(&reply)); + } + } + self.sent.lock().unwrap().push(frame); + Ok(()) + } + + async fn recv(&mut self, _timeout_ms: u32) -> Result, &'static str> { + match (self.fault, self.replies.pop_front()) { + (Fault::Unreadable, _) | (Fault::RecvFailsWhenIdle, None) => Err("connection reset"), + (_, reply) => Ok(reply), + } + } + + async fn close(&mut self) { + *self.closed.lock().unwrap() = true; + } +} + +/// `message` as the text frame the hub sends it in. +fn text_frame(message: &ServerMessage) -> Frame { + Frame::Text(serde_json::to_string(message).unwrap()) +} + +/// `module`'s component as its build left it, compiled for the board engine this build has. +fn compiled(module: &str) -> Vec { + let file = format!("et_ws_{}.wasm", module.replace('-', "_")); + let wasm = fs_err::read( + get_project_root() + .join("services/ws-modules") + .join(module) + .join("pkg") + .join(file), + ) + .unwrap(); + precompile(NATIVE_PULLEY, &wasm).unwrap() +} + +/// What running `component` against `hub` left. +/// +/// That is how the run ended, every line it logged, every frame it sent, and whether it closed the websocket. +async fn ran_on(component: &[u8], hub: Hub) -> (Result<(), BoardError>, Vec, Vec, bool) { + let lines = Arc::new(Mutex::new(Vec::new())); + let host = Host { + clock: AtomicU64::new(0), + lines: Arc::clone(&lines), + }; + let (sent, closed) = (Arc::clone(&hub.sent), Arc::clone(&hub.closed)); + let outcome = run(&board_engine().unwrap(), component, host, hub).await; + let lines = lines.lock().unwrap().clone(); + let sent = sent.lock().unwrap().clone(); + let closed = *closed.lock().unwrap(); + (outcome, lines, sent, closed) +} + +/// What running `module` to completion against a working hub left: every line it logged, and every message it sent. +async fn ran(module: &str) -> (Vec, Vec) { + let (outcome, lines, sent, closed) = ran_on(&compiled(module), Hub::default()).await; + outcome.unwrap(); + assert!(closed); + let sent = sent + .iter() + .map(|frame| { + let Frame::Text(text) = frame else { + panic!("{module} sent a frame other than text: {frame:?}"); + }; + serde_json::from_str(text).unwrap() + }) + .collect(); + assert!(lines.iter().any(|line| line.contains(&format!("agent_id={AGENT_ID}")))); + (lines, sent) +} + +/// The error `module`'s `entry.run` returned against `hub`. +async fn failed(module: &str, hub: Hub) -> EntryError { + failed_closing(module, hub).await.0 +} + +/// The error `module`'s `entry.run` returned against `hub`, and whether the websocket was closed. +async fn failed_closing(module: &str, hub: Hub) -> (EntryError, bool) { + let (outcome, _, _, closed) = ran_on(&compiled(module), hub).await; + let Err(BoardError::Guest(error)) = outcome else { + panic!("{module} did not fail: {outcome:?}"); + }; + (error, closed) +} + +#[tokio::test] +async fn data1_round_trips_a_value_through_the_boards_memory() { + let (lines, sent) = ran("wasi-data1").await; + assert!(lines.iter().any(|line| line.contains("VERIFICATION SUCCESS"))); + assert!(matches!(sent.as_slice(), [ClientMessage::Connect { agent_id: None }])); +} + +#[tokio::test] +async fn comm1_finds_itself_in_the_roster_and_broadcasts() { + let (lines, sent) = ran("wasi-comm1").await; + assert!(lines.iter().any(|line| line.contains("self present in roster"))); + assert!(matches!( + sent.as_slice(), + [ + ClientMessage::Connect { agent_id: None }, + ClientMessage::ListAgents, + ClientMessage::BroadcastMessage { .. } + ] + )); +} + +#[tokio::test] +async fn math1_sender_broadcasts_its_pointer_once_a_second_of_the_boards_clock() { + let (outcome, lines, sent, closed) = ran_on(&compiled("wasi-math1-sender"), Hub::default()).await; + outcome.unwrap(); + assert!(closed && lines.iter().any(|line| line.contains("workflow complete"))); + let [Frame::Text(connect), pointers @ ..] = sent.as_slice() else { + panic!("math1-sender did not connect first: {sent:?}"); + }; + assert!(matches!( + serde_json::from_str(connect).unwrap(), + ClientMessage::Connect { agent_id: None } + )); + assert_eq!(pointers.len(), 60); + assert!(pointers.iter().all(|pointer| matches!( + pointer, + Frame::Text(content) if content.contains(r#""bucket":"board-agent""#) + ))); +} + +#[tokio::test] +async fn math1_reads_every_message_kind_before_its_pointer_and_finds_no_input_in_a_module_bucket() { + let frames = [ + ServerMessage::AgentMessage { + message_id: "message-1".to_owned(), + from_agent_id: "agent-2".to_owned(), + scope: MessageScope::Direct, + server_received_at: "2026-11-02T06:30:00Z".to_owned(), + message: serde_json::json!({ "n": 2_i32 }), + }, + ServerMessage::MessageStatus { + message_id: None, + status: MessageDeliveryStatus::Queued, + detail: "queued".to_owned(), + }, + ServerMessage::Invalid { + message_id: None, + detail: "unknown type".to_owned(), + }, + ServerMessage::Response { + message: "pong".to_owned(), + }, + ] + .into_iter() + .map(|message| text_frame(&message)) + .chain([ + Frame::Binary(vec![7]), + Frame::Text(r#"{"type":"math1-input","bucket":"modules/wasi-math1","filename":"math1-input.json"}"#.to_owned()), + ]); + let error = failed("wasi-math1", Hub::new(Fault::None, frames)).await; + assert!(matches!(error, EntryError::Runtime(detail) if detail == "input math1-input.json not found")); +} + +#[rstest] +#[case::nothing_arrives(Fault::None)] +#[case::reads_fail(Fault::RecvFailsWhenIdle)] +#[tokio::test] +async fn math1_gives_up_when_no_pointer_arrives(#[case] fault: Fault) { + let error = failed("wasi-math1", Hub::new(fault, [])).await; + assert!(matches!(error, EntryError::Runtime(detail) if detail == "did not receive the math1-input pointer")); +} + +#[rstest] +#[case::refused(Fault::Refused, "refused", false)] +#[case::send_fails(Fault::SendFails, "broken pipe", true)] +#[case::unreadable(Fault::Unreadable, "connection reset", true)] +#[case::no_ack(Fault::NoAck, "the hub sent no et-connect-ack", true)] +#[tokio::test] +async fn data1_fails_to_connect_through_a_broken_hub( + #[case] fault: Fault, + #[case] expected: &str, + #[case] was_open: bool, +) { + let (error, closed) = failed_closing("wasi-data1", Hub::new(fault, [])).await; + assert!(matches!(error, EntryError::Ws(WsError::Transport(reason)) if reason == expected)); + assert_eq!(closed, was_open); +} + +#[tokio::test] +async fn data1_fails_to_connect_when_a_frame_claims_a_type_it_does_not_match() { + let malformed = Frame::Text(r#"{"type":"et-response"}"#.to_owned()); + let (error, closed) = failed_closing("wasi-data1", Hub::new(Fault::None, [malformed])).await; + assert!(matches!(error, EntryError::Ws(WsError::Decode(_)))); + assert!(closed); +} + +#[tokio::test] +async fn comm1_fails_when_its_roster_request_cannot_be_sent() { + let error = failed("wasi-comm1", Hub::new(Fault::SendFailsAfterConnect, [])).await; + assert!(matches!(error, EntryError::Ws(WsError::Transport(reason)) if reason == "broken pipe")); +} + +#[tokio::test] +async fn bytes_that_were_not_compiled_for_the_board_do_not_load() { + let (outcome, lines, sent, _) = ran_on(b"not a component", Hub::default()).await; + assert!(matches!(outcome, Err(BoardError::Wasmtime(_)))); + assert!(lines.is_empty() && sent.is_empty()); +} diff --git a/services/ws-wasi-runner/tests/convert.rs b/services/ws-wasi-runner/tests/convert.rs new file mode 100644 index 00000000..abd76770 --- /dev/null +++ b/services/ws-wasi-runner/tests/convert.rs @@ -0,0 +1,233 @@ +//! Covers the frame each guest `client-message` goes out as, the message each inbound frame carries, and the +//! `server-message` each of those reaches the guest as. +#![cfg(test)] + +use chrono::{DateTime, Utc}; +use edge_toolkit::ws::{ + AgentConnectionState, AgentSummary, ClientMessage, CloudEvent as EtCloudEvent, ConnectStatus, + MessageDeliveryStatus, MessageScope, ServerMessage, +}; +use et_ws_wasi_runner::bindings::et::ws_messages::messages::{ + AgentConnectionState as WitAgentConnectionState, AlivePayload, BroadcastMessagePayload, ClientEventPayload, + ClientMessage as WitClientMessage, CloudEvent, ConnectPayload, ConnectStatus as WitConnectStatus, + MessageAckPayload, MessageDeliveryStatus as WitMessageDeliveryStatus, MessageScope as WitMessageScope, + RelayBinaryPayload, RelayTextPayload, SendAgentMessagePayload, ServerMessage as WitServerMessage, +}; +use et_ws_wasi_runner::bindings::et::ws_wasi::ws::WsError; +use et_ws_wasi_runner::convert::{Frame, client_frame, server_message, server_message_to_wit}; +use rstest::rstest; +use serde_json::json; + +/// The time every event below carries. +const TIME: &str = "2026-11-02T06:30:00Z"; + +/// The text frame `message` serialises to. +fn typed(message: &ClientMessage) -> Frame { + Frame::Text(serde_json::to_string(message).unwrap()) +} + +/// A guest's event, with `time` and `specversion` as given. +fn event(time: &str, specversion: &str) -> WitClientMessage { + WitClientMessage::ClientEvent(ClientEventPayload { + event: CloudEvent { + data: r#"{"reading":1}"#.to_owned(), + id: "event-1".to_owned(), + source: "/sensor".to_owned(), + specversion: specversion.to_owned(), + time: time.to_owned(), + type_: "et.reading".to_owned(), + }, + }) +} + +#[rstest] +#[case::connect( + WitClientMessage::Connect(ConnectPayload { agent_id: Some("agent-1".to_owned()) }), + typed(&ClientMessage::Connect { agent_id: Some("agent-1".to_owned()) }), +)] +#[case::alive( + WitClientMessage::Alive(AlivePayload { timestamp: TIME.to_owned() }), + typed(&ClientMessage::Alive { timestamp: TIME.to_owned() }), +)] +#[case::list_agents(WitClientMessage::ListAgents, typed(&ClientMessage::ListAgents))] +#[case::send_agent_message( + WitClientMessage::SendAgentMessage(SendAgentMessagePayload { + message: r#"{"n":2}"#.to_owned(), + to_agent_id: "agent-2".to_owned(), + }), + typed(&ClientMessage::SendAgentMessage { to_agent_id: "agent-2".to_owned(), message: json!({ "n": 2_i32 }) }), +)] +#[case::broadcast_message( + WitClientMessage::BroadcastMessage(BroadcastMessagePayload { message: "[1,2]".to_owned() }), + typed(&ClientMessage::BroadcastMessage { message: json!([1_i32, 2_i32]) }), +)] +#[case::message_ack( + WitClientMessage::MessageAck(MessageAckPayload { message_id: "message-1".to_owned() }), + typed(&ClientMessage::MessageAck { message_id: "message-1".to_owned() }), +)] +#[case::client_event( + event(TIME, "1.0"), + typed(&ClientMessage::ClientEvent { + event: EtCloudEvent::new( + "event-1".to_owned(), + "/sensor".to_owned(), + "et.reading".to_owned(), + DateTime::parse_from_rfc3339(TIME).unwrap().with_timezone(&Utc), + json!({ "reading": 1_i32 }), + ), + }), +)] +#[case::relay_text( + WitClientMessage::RelayText(RelayTextPayload { content: "hello".to_owned() }), + Frame::Text("hello".to_owned()), +)] +#[case::relay_binary( + WitClientMessage::RelayBinary(RelayBinaryPayload { content: vec![0, 1, 2] }), + Frame::Binary(vec![0, 1, 2]), +)] +fn each_client_message_goes_out_as_its_frame(#[case] message: WitClientMessage, #[case] expected: Frame) { + assert_eq!(client_frame(message).unwrap(), expected); +} + +#[rstest] +#[case::opaque_json_that_is_not_json(WitClientMessage::BroadcastMessage(BroadcastMessagePayload { + message: "{".to_owned(), +}))] +#[case::event_time_that_is_not_rfc_3339(event("yesterday", "1.0"))] +#[case::event_specversion_no_hub_knows(event(TIME, "0.3"))] +fn a_client_message_that_cannot_be_encoded_is_a_decode_error(#[case] message: WitClientMessage) { + assert!(matches!(client_frame(message), Err(WsError::Decode(_)))); +} + +#[test] +fn a_binary_frame_is_relayed() { + assert!(matches!( + server_message(Frame::Binary(vec![7])).unwrap(), + ServerMessage::RelayBinary { content } if content == [7] + )); +} + +#[test] +fn a_text_frame_claiming_a_type_it_does_not_match_is_rejected() { + let frame = Frame::Text(r#"{"type":"et-connect-ack"}"#.to_owned()); + let _rejection = server_message(frame).unwrap_err(); +} + +#[test] +fn a_reconnect_ack_keeps_its_status() { + let ack = ServerMessage::ConnectAck { + agent_id: "agent-1".to_owned(), + status: ConnectStatus::Reconnected, + }; + assert!(matches!( + server_message_to_wit(ack).unwrap(), + WitServerMessage::ConnectAck(payload) if matches!(payload.status, WitConnectStatus::Reconnected) + )); +} + +#[test] +fn a_disconnected_agent_keeps_its_state() { + let agents = vec![AgentSummary::new( + "agent-1".to_owned(), + AgentConnectionState::Disconnected, + "192.0.2.1".parse().unwrap(), + )]; + let WitServerMessage::ListAgentsResponse(payload) = + server_message_to_wit(ServerMessage::ListAgentsResponse { agents }).unwrap() + else { + panic!("a list-agents response must stay a list-agents response"); + }; + assert!(matches!( + payload.agents.as_slice(), + [agent] if matches!(agent.state, WitAgentConnectionState::Disconnected) + )); +} + +#[rstest] +#[case::direct(MessageScope::Direct, WitMessageScope::Direct)] +#[case::broadcast(MessageScope::Broadcast, WitMessageScope::Broadcast)] +fn an_agent_message_carries_its_scope_and_payload(#[case] scope: MessageScope, #[case] expected: WitMessageScope) { + let message = ServerMessage::AgentMessage { + message_id: "message-1".to_owned(), + from_agent_id: "agent-2".to_owned(), + scope, + server_received_at: TIME.to_owned(), + message: json!({ "n": 2_i32 }), + }; + let WitServerMessage::AgentMessage(payload) = server_message_to_wit(message).unwrap() else { + panic!("an agent message must stay an agent message"); + }; + assert_eq!(payload.scope, expected); + assert_eq!(payload.message, r#"{"n":2}"#); + assert_eq!( + ( + payload.message_id.as_str(), + payload.from_agent_id.as_str(), + payload.server_received_at.as_str() + ), + ("message-1", "agent-2", TIME) + ); +} + +#[rstest] +#[case::delivered(MessageDeliveryStatus::Delivered, WitMessageDeliveryStatus::Delivered)] +#[case::queued(MessageDeliveryStatus::Queued, WitMessageDeliveryStatus::Queued)] +#[case::acknowledged(MessageDeliveryStatus::Acknowledged, WitMessageDeliveryStatus::Acknowledged)] +#[case::broadcast(MessageDeliveryStatus::Broadcast, WitMessageDeliveryStatus::Broadcast)] +fn a_message_status_carries_its_status( + #[case] status: MessageDeliveryStatus, + #[case] expected: WitMessageDeliveryStatus, +) { + let message = ServerMessage::MessageStatus { + message_id: Some("message-1".to_owned()), + status, + detail: "ok".to_owned(), + }; + let WitServerMessage::MessageStatus(payload) = server_message_to_wit(message).unwrap() else { + panic!("a message status must stay a message status"); + }; + assert_eq!(payload.status, expected); + assert_eq!( + (payload.message_id.as_deref(), payload.detail.as_str()), + (Some("message-1"), "ok") + ); +} + +#[test] +fn an_invalid_notice_carries_its_detail() { + let message = ServerMessage::Invalid { + message_id: None, + detail: "unknown type".to_owned(), + }; + assert!(matches!( + server_message_to_wit(message).unwrap(), + WitServerMessage::Invalid(payload) if payload.message_id.is_none() && payload.detail == "unknown type" + )); +} + +#[test] +fn a_response_carries_its_message() { + let message = ServerMessage::Response { + message: "pong".to_owned(), + }; + assert!(matches!( + server_message_to_wit(message).unwrap(), + WitServerMessage::Response(payload) if payload.message == "pong" + )); +} + +#[rstest] +#[case::text( + ServerMessage::RelayText { content: "hello".to_owned() }, + |message: &WitServerMessage| matches!(message, WitServerMessage::RelayText(payload) if payload.content == "hello"), +)] +#[case::binary( + ServerMessage::RelayBinary { content: vec![7] }, + |message: &WitServerMessage| matches!(message, WitServerMessage::RelayBinary(payload) if payload.content == [7]), +)] +fn a_relayed_frame_reaches_the_guest_unchanged( + #[case] message: ServerMessage, + #[case] expected: fn(&WitServerMessage) -> bool, +) { + assert!(expected(&server_message_to_wit(message).unwrap())); +} diff --git a/services/ws-wasi-runner/tests/keyvalue.rs b/services/ws-wasi-runner/tests/keyvalue.rs new file mode 100644 index 00000000..3cb1ff51 --- /dev/null +++ b/services/ws-wasi-runner/tests/keyvalue.rs @@ -0,0 +1,48 @@ +//! Covers what a `store.open` identifier opens on every host, and the `store.error` values a bucket call fails with. +#![cfg(test)] + +use et_ws_wasi_runner::bindings::wasi::keyvalue::store::Error; +use et_ws_wasi_runner::keyvalue::{Bucket, bucket_from_identifier, kv_not_implemented}; +use rstest::rstest; +use wasmtime::component::ResourceTableError; + +#[rstest] +#[case::agent("3f2c9a6e-agent", "3f2c9a6e-agent")] +#[case::agent_named_like_a_module("modules", "modules")] +fn an_agent_id_opens_its_storage_bucket(#[case] identifier: &str, #[case] expected: &str) { + let bucket = bucket_from_identifier(identifier).unwrap(); + assert!(matches!(bucket, Bucket::Storage { agent_id } if agent_id == expected)); +} + +#[rstest] +#[case::plain("modules/wasi-math1", "wasi-math1")] +#[case::scoped("modules/@edge-toolkit/wasi-math1", "@edge-toolkit/wasi-math1")] +fn a_module_name_opens_its_asset_bucket(#[case] identifier: &str, #[case] expected: &str) { + let bucket = bucket_from_identifier(identifier).unwrap(); + assert!(matches!(bucket, Bucket::Modules { module_name } if module_name == expected)); +} + +#[rstest] +#[case::empty("")] +#[case::agent_path("agent/file.json")] +#[case::no_module("modules/")] +#[case::module_path("modules/wasi-math1/pkg")] +#[case::scope_only("modules/@edge-toolkit")] +#[case::empty_scope("modules/@/wasi-math1")] +#[case::empty_scoped_name("modules/@edge-toolkit/")] +#[case::scoped_module_path("modules/@edge-toolkit/wasi-math1/pkg")] +fn an_identifier_naming_a_path_opens_nothing(#[case] identifier: &str) { + let error = bucket_from_identifier(identifier).err().unwrap(); + assert!(matches!(error, Error::Other(detail) if detail.contains(&format!("{identifier:?}")))); +} + +#[test] +fn an_unimplemented_call_names_its_operation() { + assert!(matches!(kv_not_implemented("list-keys"), Error::Other(detail) if detail == "list-keys not implemented")); +} + +#[test] +fn a_stale_bucket_handle_is_reported_as_the_table_explains_it() { + let error = Error::from(ResourceTableError::NotPresent); + assert!(matches!(error, Error::Other(detail) if detail == ResourceTableError::NotPresent.to_string())); +} diff --git a/services/ws-wasi-runner/tests/server_message_to_wit.rs b/services/ws-wasi-runner/tests/server_message_to_wit.rs index ea9f0485..5155fba5 100644 --- a/services/ws-wasi-runner/tests/server_message_to_wit.rs +++ b/services/ws-wasi-runner/tests/server_message_to_wit.rs @@ -6,7 +6,7 @@ use std::net::{IpAddr, Ipv4Addr, Ipv6Addr}; use edge_toolkit::ws::{AgentConnectionState, AgentSummary, ServerMessage}; use et_ws_wasi_runner::bindings::et::ws_messages::messages::{IpAddress, ServerMessage as WitServerMessage}; -use et_ws_wasi_runner::host::ws::server_message_to_wit; +use et_ws_wasi_runner::convert::server_message_to_wit; #[test] fn list_agents_carries_each_address_family_as_its_own_case() { diff --git a/utilities/int-gen/src/wit/bindings.rs b/utilities/int-gen/src/wit/bindings.rs index e9b98b63..5055612e 100644 --- a/utilities/int-gen/src/wit/bindings.rs +++ b/utilities/int-gen/src/wit/bindings.rs @@ -36,7 +36,7 @@ const WORLD: &str = "runner"; /// /// Mirrors the `with:` block the `bindgen!` invocation carried, and must stay in step with the runner's /// host module: a key here names a WIT item, the value the Rust path that replaces its generated form. -const WITH: [(&str, &str); 1] = [("wasi:keyvalue/store.bucket", "super::host::wasi_keyvalue::Bucket")]; +const WITH: [(&str, &str); 1] = [("wasi:keyvalue/store.bucket", "super::keyvalue::Bucket")]; /// Header prepended to the emitted file. ///