From 8587fbc5aa0f38739182fa50383bb94c8d821ed7 Mon Sep 17 00:00:00 2001 From: root Date: Sun, 14 Jun 2026 17:51:51 +0000 Subject: [PATCH] wasm runtime: central WS broker, unified handlers, W7d holdouts, membrane completeness - WS: a dedicated broker process owns HTTP_PORT + every connection; it forwards renders to the worker pool over uce.sock (non-blocking) and applies ws_* command batches flushed back at workspace teardown. Removes the now-dead per-worker websocket executor (-509 lines). - Dispatch: unify CLI / WebSocket / serve_http / page render through one serve_via_wasm(entry_unit, handler) path; handler string -> __uce_ export symbol. - W7d: rewrite zip.uce to the membrane return-value error contract (no C++ try/catch), error-reporting.uce to genuine wasm traps instead of throw, and sharedunit.uce to unit_info(); empty the native-only token gate. - Membrane: wire ls / mkdir / file_mtime through new uce_host_file_list / uce_host_file_mkdir / uce_host_file_mtime hostcalls (resolve_guest_file gains directory support). Fixes /doc/index.uce listing nothing; adds a regression assertion that the index enumerates items. - Docs: add docs/wasm-runtime-architecture.md; record the W7e staged native- deletion plan in WASM-PROPOSAL.md. Verified: scripts/run_cli_tests.sh --include-wasm-kill -> 87 passed, 0 failed. Co-Authored-By: Claude Opus 4.8 --- WASM-PROPOSAL.md | 70 ++- docs/wasm-runtime-architecture.md | 318 +++++++++++ site/demo/error-reporting.uce | 35 +- site/demo/sharedunit.uce | 8 +- site/tests/cli_runner.uce | 4 + site/tests/zip.uce | 48 +- src/lib/fcgi_forward.h | 172 ++++++ src/lib/sys.cpp | 81 ++- src/linux_fastcgi.cpp | 866 ++++++++++-------------------- src/wasm/backend.cpp | 83 ++- src/wasm/backend.h | 24 +- src/wasm/core.cpp | 101 ++-- src/wasm/core_hostcalls.syms | 3 + src/wasm/worker.cpp | 175 ++++-- 14 files changed, 1219 insertions(+), 769 deletions(-) create mode 100644 docs/wasm-runtime-architecture.md create mode 100644 src/lib/fcgi_forward.h diff --git a/WASM-PROPOSAL.md b/WASM-PROPOSAL.md index 6dcc533..64eff5f 100644 --- a/WASM-PROPOSAL.md +++ b/WASM-PROPOSAL.md @@ -1,5 +1,10 @@ # WASM-PROPOSAL: WebAssembly Unit Runtime for UCE +> **Steady-state reference:** for the runtime as it is actually built today +> (process topology, the membrane, unified dispatch, and the central WebSocket +> broker), see [`docs/wasm-runtime-architecture.md`](docs/wasm-runtime-architecture.md). +> This proposal retains the motivation and the phased migration plan. + - **Status:** building the production worker per §9.1 (W-phases). W1 (core module), W2 (wasm unit compile target), W3 (workspace runtime + membrane, `src/wasm/worker.cpp`), and W4 (config-selectable FastCGI backend, @@ -970,14 +975,63 @@ and the operational services test) longer per-request timeouts while preserving strict status/body checks, and the services test drains split custom-server response packets before deciding that the named handler failed. -> **Status: W7a/W7b PARTIAL (2026-06-13).** CLI/WebSocket/SERVE_HTTP entry kind -> plumbing exists, CLI socket error propagation is fixed, and post-fork ticker -> reset is safer. Directly enabling wasm `SERVE_HTTP` in the in-process custom -> HTTP dispatcher was tested and reverted: the dispatcher can hang/spin when it -> renders wasm in the forked broker. W7c therefore remains a design item: route -> custom-server requests back through a normal worker/workspace boundary rather -> than rendering inside the broker child. Remaining native fallback tokens after -> W6/W7 hardening are still `zip_` and direct `compiler_load_shared_unit(`. +> **Status: W7a/W7b/W7c/W7d DONE (2026-06-14).** Handler invocation is now +> unified: CLI / WebSocket / serve_http / page render all route through one +> `serve_via_wasm(entry_unit, handler)` path, the handler string mapping to a +> `__uce_` export (see +> [`docs/wasm-runtime-architecture.md`](docs/wasm-runtime-architecture.md)). +> W7c was resolved by the central WebSocket broker + serve_http dispatcher both +> forwarding to a clean worker over `uce.sock` (no wasm render inside a forked +> broker child). W7d cleared the last native-only holdouts: `zip.uce` was +> rewritten to use the membrane's return-value error contract (no C++ +> try/catch), `error-reporting.uce` now triggers genuine wasm traps instead of +> `throw`, and `sharedunit.uce` uses `unit_info()` instead of +> `compiler_load_shared_unit()`. **The native-only token gate is now empty** — +> no unit source forces native anymore. +> +> **W7e plan — deleting the native pipeline (the remaining work).** The native +> pipeline is not just a fallback; it is currently the **wasm compile driver**: +> `compiler.cpp` preprocesses `.uce → C++` (shared front-end, KEEP), then +> compiles the native `.so` (`COMPILE_SCRIPT`), and only *then*, gated on that +> native compile succeeding, compiles the `.wasm` side-module +> (`compile_wasm_unit`). The backend's native fallback on a cold/stale artifact +> exists so the native path can JIT-rebuild **both** `.so` and `.wasm`. So the +> deletion must be staged, not a single excision: +> 1. **Decouple the compile driver.** Make the `.wasm` side-module compile run +> unconditionally from the shared preprocessed C++, independent of the +> native `.so` compile and of the `SharedUnit` struct's native bookkeeping. +> 2. **Make wasm self-healing on cold/stale.** Have the backend trigger (or +> wait on) an on-demand wasm compile instead of falling back to native, so +> `wasm_artifact_exists` staleness no longer needs the native rebuild. (Note: +> until this lands, a cold request — before the proactive compiler warms a +> unit — to a unit rewritten for the membrane error contract, e.g. `zip.uce`, +> can `500` on the native path because native `zip_create` still throws.) +> 3. **Remove native execution + machinery.** Delete `compiler_invoke` / +> `_cli` / `_websocket` / `_serve_http`, `compiler_load_shared_unit`, +> `load_shared_unit`, the `dlopen`/`dlsym`/`dlclose` path and `.so` +> function-pointer fields on `SharedUnit`, the `COMPILE_SCRIPT` native +> compile, and the native-fallback branches in `handle_complete` and +> `wasm_backend_should_handle`. +> 4. **Drop the feature flag.** Remove `WASM_BACKEND_ENABLED` and the now-dead +> `wasm_backend_native_fallback_*` gate (the token scan is already vestigial). +> +> Each stage must rebuild green on `scripts/run_cli_tests.sh --include-wasm-kill` +> before the next. W7f (broad dead/legacy sweep) folds into stages 3–4. +> +> **Prerequisite — membrane silent-stub completeness (in progress).** Removing +> native fallback (stages 2–4) turns any silently-stubbed wasm-core function +> into a hard failure with no safety net. A real one slipped through this way: +> `ls()` was a stub returning empty, so `doc/index.uce` (which enumerates pages +> via `ls("pages/")`) rendered but listed nothing — and the suite missed it +> because it only asserted the page returned ``, never that it enumerated +> items. Fixed (2026-06-14) by wiring `ls`, `mkdir`, and `file_mtime` through new +> `uce_host_file_list` / `uce_host_file_mkdir` / `uce_host_file_mtime` hostcalls +> (and teaching `resolve_guest_file` to accept directories), plus a regression +> assertion that the doc index contains `func-item`. Remaining wasm-core stubs +> are intentional: `shell_exec` (no shell in the sandbox), the `*_locked_fd` +> fd-locking variants, and the `backtrace_*` / `signal_name` host-only helpers. +> Before stage 2, audit the rest of the wasm-core block in `src/lib/sys.cpp` for +> any other `(void)arg; return {}` stub a unit can reach. --- diff --git a/docs/wasm-runtime-architecture.md b/docs/wasm-runtime-architecture.md new file mode 100644 index 0000000..ac1db47 --- /dev/null +++ b/docs/wasm-runtime-architecture.md @@ -0,0 +1,318 @@ +# UCE WASM Runtime Architecture + +Status: current as of the W7 cutover (June 2026). This document describes the +**runtime architecture as built** — the process topology, the wasm membrane, +the unified request dispatch, and the central WebSocket broker. For the +historical motivation and the phased migration plan, see +[`WASM-PROPOSAL.md`](../WASM-PROPOSAL.md); this file is the steady-state +reference that proposal points at. + +The guiding principle: **request code never shares an address space or an +allocator with the runtime.** Every unit runs as a WebAssembly module inside a +per-request workspace behind a narrow host membrane. Long-lived connection +state (WebSocket connections, listening HTTP sockets) lives in dedicated native +broker processes that own no unit code — they hold the connection and forward +the actual unit invocation to a clean-engine worker, exactly like a normal +page render. + +--- + +## 1. Process topology + +A single native binary (`uce_fastcgi`) forks into a small set of long-lived +roles. None of them is special-cased per request mode; the differences are +purely *which socket a process listens on* and *which function inside a unit +gets invoked*. + +``` + ┌────────────────────────────┐ + nginx ──FastCGI──► worker pool (N processes) │ /run/uce.sock + (port 80 etc.) │ uniform unit renderers │ (FastCGI + CLI) + └─────────────▲──────────────┘ + │ forward render (FastCGI, uce.sock) + │ + browser ──raw HTTP / WS──► ┌────────┴─────────┐ + (HTTP_PORT 8080) │ WS broker │ owns HTTP_PORT + every + │ (1 process) │ WS connection; renders + └────────▲─────────┘ nothing itself + │ ws_* command flush + │ (FastCGI, ws-broker.sock) + on-demand serve_http ──► ┌────────┴─────────┐ + (per bind addr) │ custom-server │ owns one serve_http + │ dispatcher(s) │ bind addr; forwards to pool + └──────────────────┘ + + parent process: spawns/respawns all of the above + the proactive compiler. +``` + +| Process | Owns | Renders units? | Source | +|---|---|---|---| +| **Parent** | nothing; supervises children | no | `main()`, `init_base_process()` | +| **Worker** (×`WORKER_COUNT`) | `FCGI_SOCKET_PATH` (`/run/uce.sock`) + `CLI_SOCKET_PATH` | **yes** — the only processes that run wasm | `listen_for_connections()` | +| **WS broker** (×1) | `HTTP_PORT` + every live WS connection + `WS_BROKER_SOCKET_PATH` | no — forwards to the pool | `run_ws_broker()` | +| **serve_http dispatcher** (×bind) | one custom-server bind address | no — forwards to the pool | `custom_server_http_dispatcher_loop()` | +| **Proactive compiler** | nothing; pre-compiles units | no | `run_proactive_compiler()` | + +The key invariant: **only workers instantiate Wasmtime and run unit code.** +Every connection-owning process (broker, serve_http dispatcher) forwards the +real invocation back to a worker over `/run/uce.sock` using the minimal +FastCGI client in [`src/lib/fcgi_forward.h`](../src/lib/fcgi_forward.h). This is +forced by Wasmtime: an `Engine`/`Store` cannot be safely re-created across +`fork()`, and the brokers fork from the parent that already touched the +runtime. So the brokers hold the long-lived connection and units respond to +events the same way they respond to a page request — through a clean worker. + +--- + +## 2. The membrane and the DValue ABI + +Units are compiled to WebAssembly and linked against a host import surface (the +"membrane"). Host and guest exchange structured values as **UCEB** (a compact +binary `DValue` encoding — `ucb_encode`/`ucb_decode`); strings cross as +`std::string` on the native side throughout (no raw `char*` ownership across +the boundary). + +- `DValue` is the universal value type: scalars plus an ordered child map + (`_map`, a `std::map`). `operator[]` is non-const (creates); + `.key(k)` is the const probe returning `const DValue*`; `.each(fn)` iterates + children yielding `const DValue&`. +- The request context (`params`/`get`/`post`/`cookies`/`session`, the raw body + `in`, and—for WS—the connection context) is marshalled into a single `ctx` + DValue, UCEB-encoded, and handed to the workspace. The response (body, + headers, status, and any `meta` such as `ws_commands`) comes back the same + way. + +See [`docs/wasm-phase1-dvalue-abi.md`](wasm-phase1-dvalue-abi.md) for the wire +format details. + +--- + +## 3. Units, handlers, and export naming + +A `.uce` unit compiles to a wasm module that exports one symbol per **handler**. +There is no per-mode machinery — a "CLI unit", "WebSocket unit", and "page" are +the same compiled artifact invoked at different exports. The dispatcher passes a +**handler string**; the host maps it to an export symbol: + +``` +__uce_[_] +``` + +`handler_export_symbol()` (`src/wasm/worker.cpp`) splits on the first `:`: + +| Handler string | Export symbol | +|---|---| +| `render` | `__uce_render` | +| `cli` | `__uce_cli` | +| `websocket` | `__uce_websocket` | +| `once` | `__uce_once` (optional; absence is not an error) | +| `serve_http` | `__uce_serve_http` | +| `serve_http:named` | `__uce_serve_http_named` | +| `component:CARD` | `__uce_component_CARD` | +| `exists` | probe only — resolves the unit, loads nothing | + +`sanitize_symbol_suffix()` keeps `[A-Za-z0-9_]` (mirrors `ascii_safe_name`). +`wasm_resolve_target(unit, handler)` (`src/wasm/core.cpp`) resolves the source +path and looks up the export's funcref slot; `exists` lets callers probe a unit +without instantiating it. + +`wasm_backend_should_handle(request, entry_unit)` gates whether the wasm path +takes the request (vs. the legacy native compiler path, still selectable while +the native pipeline is retired). With `WASM_BACKEND_ENABLED=1` this is the +default for all unit execution. + +--- + +## 4. The workspace runtime + +Each request gets a fresh **workspace** — a per-request wasm instance tree with +the membrane wired in. `wasm_worker_serve(worker, ctx, entry_unit, handler)` +(`src/wasm/worker.cpp`) is the single entry point for *every* mode: + +1. Birth a workspace (CoW-snapshot-based where available). +2. Resolve `entry_unit` + `handler` to an export; components referenced at + runtime are resolved on demand via the `uce_host_component_resolve` hostcall + (`component_resolve()` → `__uce_<...>` slot), loading dependency modules + lazily and recording resolve counts/timings. +3. Invoke the export. Unit code calls host functions (filesystem, sqlite, regex, + markdown, time, tasks, `ws_*`, …) across the membrane. +4. Collect the response (`WasmResponse`: body, headers, status, `meta`). + +Because the workspace owns no process-lifetime state, a unit crash or trap is +contained: it fails the one request and the worker stays healthy. Workers +suspend the native SIGSEGV/SIGILL recovery handler around the wasm call so that +Wasmtime's own trap signals are not escalated into a native fatal signal (see +`serve_via_wasm` in `handle_complete`). + +--- + +## 5. Request dispatch (`handle_complete`) + +`handle_complete()` (`src/linux_fastcgi.cpp`) is the worker's single dispatch +point. It selects a handler string from request params and calls the same +`serve_via_wasm(entry_unit, handler)` lambda for all of them: + +``` +UCE_WS == "1" → serve_via_wasm(entry_unit, "websocket") +request.resources.is_cli → serve_via_wasm(entry_unit, "cli") +UCE_SERVE_HTTP == "1" → serve_via_wasm(entry_unit, "serve_http"[:fn]) +otherwise (page) → serve_via_wasm(entry_unit, "render") +``` + +The `UCE_*` params are set by whichever broker forwarded the request: + +- **Page render**: nginx → `/run/uce.sock` directly; no `UCE_*` flags → `render`. +- **CLI**: the CLI socket sets `is_cli`. +- **serve_http**: the custom-server dispatcher sets `UCE_SERVE_HTTP=1` plus + `UCE_SERVE_HTTP_FUNCTION` and rewrites `SCRIPT_FILENAME` to the configured + unit (`custom_server_http_complete`). +- **WebSocket**: the WS broker sets `UCE_WS=1` and carries the connection + identity as `UCE_WS_*` params (see §6). `handle_complete` rebuilds + `request.resources.websocket_*` and `request.connection` from them before + invoking `__uce_websocket`. + +Each branch has a native-compiler fallback (`compiler_invoke*`) used only when +`wasm_backend_should_handle` declines — the holdout path during native +retirement. + +--- + +## 6. The central WebSocket broker + +The broker is the architectural centerpiece. **One process owns the HTTP port +and every WebSocket connection**, so any unit's `ws_*` call can reach any or all +connections, and a unit-code crash (which happens in a worker) never drops live +connections. + +### 6.1 Why central, and why it forwards + +Earlier designs gave each worker its connections and captured `ws_*` output to +return inline. That was wrong: connections couldn't outlive a worker, and one +unit could only talk to connections it happened to own. The broker fixes both — +it is the sole connection registry and data broker between connected clients and +the units that handle them. It renders nothing itself; like every other +connection owner it forwards unit invocation to the worker pool. + +### 6.2 Inbound: a WS frame → a worker render (non-blocking) + +`ws_broker_ws_message(request, message, opcode)` fires when a complete +(reassembled) WS message arrives on a connection. It does **not** block the +broker loop: + +1. Build FastCGI params: `SCRIPT_FILENAME`, `REQUEST_METHOD=GET`, a + `REQUEST_URI` (required — `handle_request()` rejects requests without one + *before* `on_complete` runs), `UCE_WS=1`, and the connection context as + `UCE_WS_CONNECTION_ID / SCOPE / OPCODE / BINARY / CONNECTIONS / STATE`. +2. The message rides as `UCE_WS_MESSAGE` (base64) with an **empty STDIN body** — + a non-empty STDIN makes the FastCGI transport flush a premature response + before `on_complete` ever runs. +3. Connect to `/run/uce.sock` (non-blocking) and queue the encoded request in + `ws_broker_outbound[fd]`. + +`ws_broker_drain_outbound()` runs after every `process(50)` tick: it finishes +writing each queued request, then drains and discards the reply (the unit's +output comes back via the command socket, not this reply), closing the fd when +the worker closes its end. + +### 6.3 Outbound: `ws_*` commands flushed back to the broker + +Any unit code — not just WebSocket handlers — may call `ws_send` / `ws_send_to` +/ `ws_close`. In the workspace these **record dispatch commands** rather than +touching a socket (the workspace owns no connections); `wasm-core`'s `ws_*` +(`src/lib/sys.cpp`) append to `websocket_dispatch_commands`, and +`finish_response_meta` (`src/wasm/core.cpp`) emits them as `ws_commands` (plus +`ws_connection_state` if the handler mutated per-connection state). + +`wasm_backend_serve` (`src/wasm/backend.cpp`) flushes that batch at workspace +teardown — in **any** scenario, not just WS handlers — to the broker's command +socket (`WS_BROKER_SOCKET_PATH`, `/run/uce/ws-broker.sock`) via +`fcgi_forward_request` with `UCE_WS_DISPATCH=1`. This is the only path WS data +takes out of a workspace. + +`ws_broker_apply_commands()` decodes the batch and applies each command against +the full registry it owns: `broadcast` (by scope), `send_to` (by connection id), +`close`. If the batch carries `connection_state`, it persists that onto the +matching live connection's `websocket_state`. + +### 6.4 Un-upgraded HTTP on the WS port + +The WS port can also receive ordinary (non-Upgrade) HTTP requests. +`ws_broker_complete()` routes by param: `UCE_WS_DISPATCH=1` → apply commands; +otherwise → `forward_request_to_worker()` — the *same* shared facility the +serve_http dispatcher uses, so there is no duplicated request-forwarding code. + +### 6.5 The broker loop + +`run_ws_broker()` drops the worker listeners it inherited +(`close_inherited_server_sockets`), installs permissive `on_request`/`on_data` +handlers (it renders nothing, so it accepts every request straight through to +`on_complete`), wires `on_complete=ws_broker_complete` and +`on_websocket_message=ws_broker_ws_message`, listens on `HTTP_PORT` and the +command socket, and loops `process(50)` + `drain_outbound()`. The design is +**non-blocking outbound dispatch + async command-socket flush, all in the +broker's single epoll loop** — the broker never blocks on a worker. + +The parent respawns the broker if it dies (`ws_broker_alive` / `ensure_ws_broker` +in `main()`). A broker restart loses live connections (acceptable: crashes +happen in workers, so the broker stays up in practice), but no unit state is at +risk because the broker holds none. + +--- + +## 7. The serve_http facility + +`serve_http` units are reachable on their own bind address via a custom-server +dispatcher (`custom_server_http_dispatcher_loop`). The dispatcher owns the bind +socket and, on each request, sets `UCE_SERVE_HTTP=1` + the configured unit/ +function and calls `forward_request_to_worker()`. The worker then runs +`serve_via_wasm(entry_unit, "serve_http"[:fn])`. This is the same +hold-connection-forward-render model as the WS broker, sharing +`forward_request_to_worker` and `fcgi_forward.h`. + +--- + +## 8. Build / object layout + +The native side is split into separately-compiled objects so editing the wasm +runtime does not recompile the whole TU (`scripts/build_linux.sh`): + +| Object | Contents | +|---|---| +| `bin/sqlite3.o` | sqlite amalgamation (cached) | +| `bin/wasm.o` | `src/wasm/wasm_module.cpp` (backend.cpp + worker.cpp + wasmtime) | +| `bin/main.o` | `src/linux_fastcgi.cpp` (includes `lib/uce_lib.cpp`) | + +Linked into one `-rdynamic` binary. ODR hazards across the objects are handled +deliberately: `context` and `my_pid`/`parent_pid` are `extern` (guarded for the +wasm core vs. unit builds), `operator new`/`delete` live in `types.cpp`, and +header free-functions are `inline`. The wasm backend exposes only declarations +(`src/wasm/backend.h`) to `main.o`. + +--- + +## 9. Configuration keys + +| Key | Default | Meaning | +|---|---|---| +| `WASM_BACKEND_ENABLED` | `1` | Route unit execution through the wasm path. | +| `WASM_BACKEND_VERBOSE` | `0` | Emit `X-UCE-Wasm-*` workspace timing headers (benchmark only). | +| `FCGI_SOCKET_PATH` | `/run/uce.sock` | Worker pool FastCGI socket (brokers forward here). | +| `CLI_SOCKET_PATH` | `/run/uce/cli.sock` | Worker CLI socket. | +| `HTTP_PORT` | `8080` | Raw HTTP + WebSocket port — owned by the WS broker. | +| `WS_BROKER_SOCKET_PATH` | `/run/uce/ws-broker.sock` | Broker command socket for `ws_*` flushes. | +| `WORKER_COUNT` | `4` | Number of uniform worker processes. | + +--- + +## 10. Testing + +- **Regression gate**: `scripts/run_cli_tests.sh --include-wasm-kill` runs the + in-runtime CLI suite (`site/tests/cli_runner.uce`) plus the site test pages. + Current baseline: **86 passed, 0 failed**. +- **WebSocket end-to-end**: a headless client performs a raw WS handshake to + `:HTTP_PORT` with path `/site/tests/websockets.ws.uce` (self-resolving + `SCRIPT_FILENAME`) and asserts the `hello-ack` frame — exercising the full + broker → worker → broker → client chain across process boundaries. + +All builds, runs, and installs happen on the dev host (`k-uce` / uce-dev) over +SSH; the local checkout is edit-only. diff --git a/site/demo/error-reporting.uce b/site/demo/error-reporting.uce index 89cce27..c3bfed1 100644 --- a/site/demo/error-reporting.uce +++ b/site/demo/error-reporting.uce @@ -1,5 +1,13 @@ #include "demo_guard.h" +// Fault-injection demo. Under the wasm runtime every unit fault is a guest +// trap that the workspace turns into a clean 500 without harming the worker; +// these modes exercise the three distinct trap causes (mirrors tests/wasm-kill). +u64 error_reporting_recurse(volatile u64 depth) +{ + volatile u64 next = depth + 1; + return(next + error_reporting_recurse(next)); +} RENDER(Request& context) { @@ -10,12 +18,19 @@ RENDER(Request& context) } String mode = context.get["mode"]; - if(mode == "exception") - throw std::runtime_error("Intentional test exception from /test/error-reporting.uce"); - if(mode == "abort") - raise(SIGABRT); - if(mode == "segfault") - raise(SIGSEGV); + if(mode == "trap") + __builtin_trap(); + if(mode == "recurse") + { + volatile u64 sink = error_reporting_recurse(0); + (void)sink; + } + if(mode == "loop") + { + volatile u64 i = 0; + while(i >= 0) + i++; + } <> @@ -23,11 +38,11 @@ RENDER(Request& context) UCE Test: Error reporting -

These actions intentionally trigger failures so you can verify that UCE returns a usable `500` response instead of dropping the upstream connection.

+

These actions intentionally trigger guest traps so you can verify that UCE returns a usable `500` response from a healthy worker instead of dropping the upstream connection.

} diff --git a/site/demo/sharedunit.uce b/site/demo/sharedunit.uce index 5e0162b..5593077 100644 --- a/site/demo/sharedunit.uce +++ b/site/demo/sharedunit.uce @@ -1,7 +1,9 @@ RENDER(Request& context) { - auto p = compiler_load_shared_unit(&context, "post.uce"); - if(p) - print(to_string(p)); + // The native SharedUnit loader has been retired. Unit metadata is now + // exposed through the wasm membrane via unit_info() (uce_host_units), which + // resolves and introspects another unit without dlopen-ing a native .so. + DValue info = unit_info("post.uce"); + print(json_encode(info)); } diff --git a/site/tests/cli_runner.uce b/site/tests/cli_runner.uce index 1d1c1c4..23c9a32 100644 --- a/site/tests/cli_runner.uce +++ b/site/tests/cli_runner.uce @@ -155,6 +155,10 @@ void cli_run_demo_smoke() void cli_run_http_smoke() { cli_expect_http("uce_http_smoke:doc index", "/doc/index.uce", 200, ""); + // Regression guard: the index enumerates pages via ls("pages/"). When that + // returned empty (wasm ls() was a stub), the page still rendered but + // listed nothing — so assert the function grid is actually populated. + cli_expect_http("uce_http_smoke:doc index lists functions", "/doc/index.uce", 200, "func-item"); cli_expect_http("uce_http_smoke:doc singlepage", "/doc/singlepage.uce", 200, ""); cli_expect_http("uce_http_smoke:doc component page", "/doc/index.uce?p=component", 200, "component()"); cli_expect_http("uce_http_smoke:doc regex page", "/doc/index.uce?p=regex_search", 200, "regex_search"); diff --git a/site/tests/zip.uce b/site/tests/zip.uce index 7fae3b9..5674714 100644 --- a/site/tests/zip.uce +++ b/site/tests/zip.uce @@ -49,33 +49,19 @@ RENDER(Request& context) check("zip_read()", hello == "Hello ZIP", hello); check("zip_extract()", extracted && nested == "Nested file" && value == "42", nested + " / " + value); - bool unsafe_rejected = false; - try - { - DValue unsafe_entries; - unsafe_entries["/absolute.txt"] = "bad"; - zip_create(path_join(base, "unsafe.zip"), unsafe_entries); - } - catch(std::exception& e) - { - unsafe_rejected = contains(e.what(), "unsafe"); - } + // Unsafe member names are rejected host-side; the membrane reports the + // failure as a false return (no thrown exception crosses into the unit). + DValue unsafe_entries; + unsafe_entries["/absolute.txt"] = "bad"; + bool unsafe_rejected = !zip_create(path_join(base, "unsafe.zip"), unsafe_entries); check("zip_create() rejects unsafe names", unsafe_rejected, "absolute member name rejected"); - bool nul_name_rejected = false; - try - { - DValue unsafe_entries; - String nul_name = "prefix"; - nul_name.push_back((char)0x00); - nul_name += "suffix.txt"; - unsafe_entries[nul_name] = "bad"; - zip_create(path_join(base, "nul-name.zip"), unsafe_entries); - } - catch(std::exception& e) - { - nul_name_rejected = contains(e.what(), "unsafe"); - } + DValue nul_entries; + String nul_name = "prefix"; + nul_name.push_back((char)0x00); + nul_name += "suffix.txt"; + nul_entries[nul_name] = "bad"; + bool nul_name_rejected = !zip_create(path_join(base, "nul-name.zip"), nul_entries); check("zip_create() rejects NUL entry names", nul_name_rejected, "embedded NUL member name rejected"); String binary_source("UCE", 3); @@ -98,15 +84,9 @@ RENDER(Request& context) check("gz_compress()", gz_body.size() > gz_source.size() && (u8)gz_body[0] == 0x1f && (u8)gz_body[1] == 0x8b, "bytes=" + std::to_string((u64)gz_body.size())); check("gz_uncompress()", gz_roundtrip == gz_source && gz_binary_roundtrip == binary_source && gz_binary_roundtrip.size() == binary_source.size(), "text=" + gz_roundtrip + ", binary bytes=" + std::to_string((u64)gz_binary_roundtrip.size())); - bool bad_gz_rejected = false; - try - { - gz_uncompress("not gzip data"); - } - catch(std::exception& e) - { - bad_gz_rejected = contains(e.what(), "gz_uncompress"); - } + // Invalid gzip input fails host-side; the membrane returns an empty result + // rather than throwing into the unit. + bool bad_gz_rejected = gz_uncompress("not gzip data") == ""; check("gz_uncompress() rejects invalid data", bad_gz_rejected, "invalid stream rejected"); site_tests_summary(passed, failed, skipped, "ZIP tests write only under /tmp/uce-site-tests-zip."); diff --git a/src/lib/fcgi_forward.h b/src/lib/fcgi_forward.h new file mode 100644 index 0000000..71398e1 --- /dev/null +++ b/src/lib/fcgi_forward.h @@ -0,0 +1,172 @@ +#pragma once + +// Minimal FastCGI client used by the connection brokers (the custom HTTP server +// dispatcher and the websocket exec child) to render a request through a normal +// worker on /run/uce.sock instead of rendering wasm in the broker's own forked +// process. Wasmtime cannot be safely re-created across fork, so the broker owns +// the connection but forwards the actual unit invocation to a clean-engine +// worker — the "broker holds connections, units respond like RENDER()" model. +// +// Native-only; compiled into the main object. Needs String/StringMap (uce_lib.h) +// plus the socket headers below. + +#include +#include +#include +#include +#include + +struct FcgiForwardResult { + bool ok = false; + String error; + int status = 200; // parsed from a CGI "Status:" header if present + StringMap headers; // response headers (excluding Status) + String body; // response body +}; + +// One FastCGI stream (PARAMS/STDIN): data records (<=64KB each) then the +// terminating empty record the protocol requires to close the stream. +inline void fcgi_forward_stream(String& out, unsigned char type, const String& content) +{ + size_t off = 0, len = content.size(); + while(off < len) + { + size_t chunk = std::min(len - off, (size_t)0xffff); + unsigned char hdr[8] = { 1, type, 0, 1, (unsigned char)((chunk >> 8) & 0xff), (unsigned char)(chunk & 0xff), 0, 0 }; + out.append((const char*)hdr, 8); + out.append(content.data() + off, chunk); + off += chunk; + } + unsigned char term[8] = { 1, type, 0, 1, 0, 0, 0, 0 }; + out.append((const char*)term, 8); +} + +// FastCGI name/value pair length prefix: 1 byte if < 128, else 4 bytes (high bit set). +inline void fcgi_forward_put_len(String& out, size_t n) +{ + if(n < 128) + out.push_back((char)(unsigned char)n); + else + { + out.push_back((char)(unsigned char)(((n >> 24) & 0xff) | 0x80)); + out.push_back((char)(unsigned char)((n >> 16) & 0xff)); + out.push_back((char)(unsigned char)((n >> 8) & 0xff)); + out.push_back((char)(unsigned char)(n & 0xff)); + } +} + +// Build the full FastCGI request bytes (BEGIN_REQUEST + PARAMS + STDIN) for a +// RESPONDER request id 1. The broker reuses this to fire renders non-blocking. +inline String fcgi_build_request(const StringMap& params, const String& stdin_body) +{ + String request; + unsigned char begin[16] = { 1, 1, 0, 1, 0, 8, 0, 0, 0, 1, 0, 0, 0, 0, 0, 0 }; + request.append((const char*)begin, 16); + String params_encoded; + for(auto& kv : params) + { + fcgi_forward_put_len(params_encoded, kv.first.size()); + fcgi_forward_put_len(params_encoded, kv.second.size()); + params_encoded.append(kv.first); + params_encoded.append(kv.second); + } + fcgi_forward_stream(request, 4 /*FCGI_PARAMS*/, params_encoded); + fcgi_forward_stream(request, 5 /*FCGI_STDIN*/, stdin_body); + return(request); +} + +// Forward `params` + `stdin_body` to the FastCGI responder at unix `socket_path` +// and return the parsed CGI response. Times out (recv) at `timeout_seconds`. +inline FcgiForwardResult fcgi_forward_request(const String& socket_path, + const StringMap& params, const String& stdin_body, u32 timeout_seconds = 30) +{ + FcgiForwardResult result; + + int fd = ::socket(AF_UNIX, SOCK_STREAM, 0); + if(fd < 0) + { + result.error = "fcgi_forward: socket() failed"; + return(result); + } + struct sockaddr_un addr; + memset(&addr, 0, sizeof(addr)); + addr.sun_family = AF_UNIX; + strncpy(addr.sun_path, socket_path.c_str(), sizeof(addr.sun_path) - 1); + if(::connect(fd, (struct sockaddr*)&addr, sizeof(addr)) < 0) + { + ::close(fd); + result.error = "fcgi_forward: connect(" + socket_path + ") failed"; + return(result); + } + struct timeval tv; + tv.tv_sec = timeout_seconds; + tv.tv_usec = 0; + setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, (const char*)&tv, sizeof(tv)); + + String request = fcgi_build_request(params, stdin_body); + if(::send(fd, request.data(), request.size(), MSG_NOSIGNAL) != (ssize_t)request.size()) + { + ::close(fd); + result.error = "fcgi_forward: short write to responder"; + return(result); + } + + // Read records; collect FCGI_STDOUT content until FCGI_END_REQUEST or EOF. + String inbuf, stdout_data; + char buf[65536]; + bool ended = false; + while(!ended) + { + ssize_t n = ::recv(fd, buf, sizeof(buf), 0); + if(n <= 0) + break; + inbuf.append(buf, n); + while(inbuf.size() >= 8) + { + const unsigned char* h = (const unsigned char*)inbuf.data(); + size_t content = ((size_t)h[4] << 8) + h[5]; + size_t padding = h[6]; + size_t record_len = 8 + content + padding; + if(inbuf.size() < record_len) + break; + unsigned char type = h[1]; + if(type == 6 /*FCGI_STDOUT*/) + stdout_data.append(inbuf.data() + 8, content); + else if(type == 3 /*FCGI_END_REQUEST*/) + ended = true; + inbuf.erase(0, record_len); + } + } + ::close(fd); + + // Parse the CGI response: header block, then body. Status: header (if any) + // sets the HTTP status; everything else is a response header. + size_t sep = stdout_data.find("\r\n\r\n"); + size_t sep_len = 4; + if(sep == String::npos) + { + sep = stdout_data.find("\n\n"); + sep_len = 2; + } + String header_block = sep == String::npos ? String() : stdout_data.substr(0, sep); + result.body = sep == String::npos ? stdout_data : stdout_data.substr(sep + sep_len); + + for(String line : split(header_block, "\n")) + { + line = trim(line); + if(line == "") + continue; + size_t colon = line.find(":"); + if(colon == String::npos) + continue; + String name = trim(line.substr(0, colon)); + String value = trim(line.substr(colon + 1)); + if(to_lower(name) == "status") + result.status = (int)int_val(value); + else + result.headers[name] = value; + } + + result.ok = true; + return(result); +} diff --git a/src/lib/sys.cpp b/src/lib/sys.cpp index a478b30..60ea731 100644 --- a/src/lib/sys.cpp +++ b/src/lib/sys.cpp @@ -2,6 +2,7 @@ #include #include "types.h" #include "functionlib.h" +#include "uri.h" #include "sys.h" extern "C" { @@ -17,6 +18,9 @@ int uce_host_file_exists(const char* path, size_t path_len, const char* current, size_t uce_host_file_read(const char* path, size_t path_len, const char* current, size_t current_len, char* buf, size_t cap); int uce_host_file_write(const char* path, size_t path_len, const char* current, size_t current_len, const char* content, size_t content_len, int append); void uce_host_file_unlink(const char* path, size_t path_len, const char* current, size_t current_len); +size_t uce_host_file_list(const char* path, size_t path_len, const char* current, size_t current_len, char* buf, size_t cap); +int uce_host_file_mkdir(const char* path, size_t path_len, const char* current, size_t current_len); +int64_t uce_host_file_mtime(const char* path, size_t path_len, const char* current, size_t current_len); int uce_host_task_spawn(const char* key, size_t key_len, uint64_t callback_id, double interval, uint64_t timeout, int repeat); int uce_host_task_pid(const char* key, size_t key_len); int uce_host_task_kill(int pid, int sig); @@ -43,7 +47,11 @@ String dirname(String fn) { auto pos = fn.find_last_of('/'); return(pos == Strin String path_join(String base, String child) { if(base == "") return(child); if(child == "") return(base); if(child[0] == '/') return(child); return(base + (base.back() == '/' ? "" : "/") + child); } String path_real(String path) { return(path); } bool path_is_within(String path, String root) { return(str_starts_with(path, root)); } -bool mkdir(String path) { (void)path; return(false); } +bool mkdir(String path) +{ + String current = wasm_current_unit_file(); + return(uce_host_file_mkdir(path.data(), path.size(), current.data(), current.size()) != 0); +} bool file_exists(String path) { String current = wasm_current_unit_file(); @@ -78,14 +86,30 @@ bool file_append_contents(String file_name, String content) String cwd_get() { return("/"); } void cwd_set(String path) { (void)path; } String process_start_directory() { return("/"); } -time_t file_mtime(String file_name) { (void)file_name; return(0); } +time_t file_mtime(String file_name) +{ + String current = wasm_current_unit_file(); + return((time_t)uce_host_file_mtime(file_name.data(), file_name.size(), current.data(), current.size())); +} void file_unlink(String file_name) { String current = wasm_current_unit_file(); uce_host_file_unlink(file_name.data(), file_name.size(), current.data(), current.size()); } String expand_path(String path, String relative_to_path) { return(path_join(relative_to_path, path)); } -StringList ls(String dir) { (void)dir; return(StringList()); } +StringList ls(String dir) +{ + String current = wasm_current_unit_file(); + size_t required = uce_host_file_list(dir.data(), dir.size(), current.data(), current.size(), 0, 0); + if(required == 0) + return(StringList()); + String listing(required, 0); + size_t got = uce_host_file_list(dir.data(), dir.size(), current.data(), current.size(), &listing[0], required); + listing.resize(got <= required ? got : 0); + if(listing == "") + return(StringList()); + return(split(listing, "\n")); +} u64 config_map_u64(StringMap& cfg, String key, u64 fallback) { String raw = first(cfg[key], std::to_string(fallback)); char* end = 0; unsigned long long v = strtoull(raw.c_str(), &end, 10); return(end && *end == 0 ? (u64)v : fallback); } f64 config_map_f64(StringMap& cfg, String key, f64 fallback) { String raw = first(cfg[key], std::to_string(fallback)); char* end = 0; double v = strtod(raw.c_str(), &end); return(end && *end == 0 ? (f64)v : fallback); } bool config_bool_value(String raw, bool fallback) { if(raw == "") return(fallback); return(raw != "0" && raw != "false" && raw != "no" && raw != "off"); } @@ -169,11 +193,49 @@ String ws_connection_id() { return(context ? context->resources.websocket_connec String ws_scope() { return(context ? context->resources.websocket_scope : ""); } u8 ws_opcode() { return(context ? context->resources.websocket_opcode : 0); } bool ws_is_binary() { return(context && context->resources.websocket_is_binary); } -StringList ws_connections(String scope) { (void)scope; return(StringList()); } -u64 ws_connection_count(String scope) { (void)scope; return(0); } -bool ws_send(String message, bool binary, String scope) { (void)message; (void)binary; (void)scope; return(false); } -bool ws_send_to(String connection_id, String message, bool binary) { (void)connection_id; (void)message; (void)binary; return(false); } -bool ws_close(String connection_id) { (void)connection_id; return(false); } +StringList ws_connections(String scope) { (void)scope; return(context ? context->resources.websocket_scope_connection_ids : StringList()); } +u64 ws_connection_count(String scope) { return(ws_connections(scope).size()); } +// The wasm workspace owns no connections; ws_send/ws_close record dispatch +// commands (same shape as the native websocket_exec capture) that the host +// carries back to the broker, which sends them over the real connections. +bool ws_send(String message, bool binary, String scope) +{ + if(!context) + return(false); + DValue command; + command["action"] = "broadcast"; + command["binary"].set_bool(binary); + command["message_b64"] = base64_encode(message); + command["scope"] = scope; + context->resources.websocket_dispatch_commands.push(command); + return(true); +} +bool ws_send_to(String connection_id, String message, bool binary) +{ + if(!context) + return(false); + DValue command; + command["action"] = "send_to"; + command["binary"].set_bool(binary); + command["message_b64"] = base64_encode(message); + command["connection_id"] = connection_id; + context->resources.websocket_dispatch_commands.push(command); + return(true); +} +bool ws_close(String connection_id) +{ + if(!context) + return(false); + if(connection_id == "") + connection_id = ws_connection_id(); + DValue command; + command["action"] = "close"; + command["connection_id"] = connection_id; + command["status_code"] = (f64)1000; + command["reason"] = ""; + context->resources.websocket_dispatch_commands.push(command); + return(true); +} String backtrace_frames_string(void* const* frames, size_t size, u32 skip_frames) { (void)frames; (void)size; (void)skip_frames; return(""); } String capture_backtrace_string(u32 max_frames, u32 skip_frames) { (void)max_frames; (void)skip_frames; return(""); } String signal_name(int sig) { (void)sig; return(""); } @@ -1266,6 +1328,9 @@ StringMap make_server_settings() cfg["CONTENT_TYPE"] = "text/html; charset=utf-8"; cfg["FCGI_SOCKET_PATH"] = "/run/uce.sock"; cfg["CLI_SOCKET_PATH"] = "/run/uce/cli.sock"; + // Command socket the WS broker listens on; workers flush ws_* dispatch + // command batches here at workspace teardown. + cfg["WS_BROKER_SOCKET_PATH"] = "/run/uce/ws-broker.sock"; cfg["TMP_UPLOAD_PATH"] = "/tmp/uce/uploads"; cfg["SESSION_PATH"] = "/tmp/uce/sessions"; cfg["COMPILER_SYS_PATH"] = "."; diff --git a/src/linux_fastcgi.cpp b/src/linux_fastcgi.cpp index 81a25b1..e86eda8 100644 --- a/src/linux_fastcgi.cpp +++ b/src/linux_fastcgi.cpp @@ -3,8 +3,9 @@ // main object only needs its declarations, so editing the wasm runtime no // longer recompiles the whole native TU. #include "wasm/backend.h" +// Minimal FastCGI client: connection brokers forward to a clean-engine worker. +#include "lib/fcgi_forward.h" #include -#include #include #include #include @@ -14,10 +15,15 @@ ServerState server_state; #include "fastcgi/src/fcgicc.cc" FastCGIServer server; -pid_t http_worker_pid = 0; pid_t proactive_compiler_pid = 0; -pid_t websocket_exec_pid = 0; -bool worker_accepts_http = false; + +// The central WS broker process: owns the WS port + every connection, forwards +// renders to the worker pool over uce.sock, and applies ws_* commands flushed +// back from workers. ws_broker_outbound holds in-flight async render +// connections (fd -> bytes still to write; empty value means draining the reply). +FastCGIServer ws_broker; +pid_t ws_broker_pid = 0; +std::map ws_broker_outbound; static sigjmp_buf request_fault_jmp; static volatile sig_atomic_t request_fault_active = 0; static volatile sig_atomic_t request_fault_signal = 0; @@ -26,11 +32,6 @@ static Request* request_fault_request = 0; // captured here and symbolized after the siglongjmp. static void* request_fault_frames[64]; static volatile sig_atomic_t request_fault_frame_count = 0; -static int websocket_exec_fd = -1; -static String websocket_exec_read_buffer = ""; -static std::deque websocket_exec_pending_jobs; -static DValue websocket_exec_inflight_job; -static String websocket_exec_write_buffer = ""; void close_inherited_server_sockets(); u64 request_seed_from_time(f64 time_value); @@ -162,471 +163,6 @@ void restore_request_fault_handlers() signal(SIGALRM, on_segfault); } -namespace { - -bool websocket_ipc_set_nonblocking(int fd) -{ - int flags = fcntl(fd, F_GETFL, 0); - if(flags == -1) - return(false); - return(fcntl(fd, F_SETFL, flags | O_NONBLOCK) == 0); -} - -bool websocket_exec_enabled_for_process() -{ - return(worker_accepts_http); -} - -u64 websocket_exec_queue_limit_bytes() -{ - u64 configured = int_val(server_state.config["WEBSOCKET_EXEC_QUEUE_BYTES"]); - if(configured < 64 * 1024) - configured = 1024 * 1024; - return(configured); -} - -bool websocket_exec_has_inflight_job() -{ - return(websocket_exec_inflight_job.to_bool()); -} - -FastCGIServer::Connection* websocket_find_connection(String connection_id) -{ - for(auto& item : server.client_sockets) - { - FastCGIServer::Connection* connection = item.second; - if(connection->is_websocket && connection->websocket_connection_id == connection_id) - return(connection); - } - return(0); -} - -void websocket_exec_clear_ipc_state() -{ - if(websocket_exec_fd != -1) - close(websocket_exec_fd); - websocket_exec_fd = -1; - websocket_exec_read_buffer = ""; - websocket_exec_write_buffer = ""; - websocket_exec_inflight_job.clear(); -} - -void websocket_exec_close_connection(String connection_id, u16 status_code = 1011, String reason = "websocket handler unavailable") -{ - if(connection_id == "") - return; - server.websocket_close(connection_id, status_code, reason); -} - -void websocket_exec_fail_inflight_job(String reason = "websocket handler unavailable") -{ - if(!websocket_exec_has_inflight_job()) - return; - websocket_exec_close_connection(websocket_exec_inflight_job["connection_id"].to_string(), 1011, reason); - websocket_exec_inflight_job.clear(); - websocket_exec_write_buffer = ""; -} - -void websocket_exec_queue_job(DValue job) -{ - if(job["connection_id"].to_string() == "") - return; - - u64 queued_bytes = websocket_exec_write_buffer.length(); - if(websocket_exec_has_inflight_job()) - queued_bytes += websocket_exec_inflight_job["serialized"].to_string().length(); - for(auto& pending : websocket_exec_pending_jobs) - queued_bytes += pending["serialized"].to_string().length(); - queued_bytes += json_encode(job).length(); - - if(queued_bytes > websocket_exec_queue_limit_bytes()) - { - printf("(!) websocket dispatch queue overflow for %s\n", job["connection_id"].to_string().c_str()); - websocket_exec_close_connection(job["connection_id"].to_string(), 1013, "websocket server busy"); - return; - } - - job["serialized"] = json_encode(job) + "\n"; - websocket_exec_pending_jobs.push_back(job); -} - -StringList websocket_exec_snapshot_connections(String scope) -{ - return(server.websocket_connection_ids(scope)); -} - -void websocket_exec_append_command(DValue command) -{ - if(!context) - return; - context->resources.websocket_dispatch_commands.push(command); -} - -DValue websocket_exec_make_message_command(String action, String message, bool binary) -{ - DValue command; - command["action"] = action; - command["binary"].set_bool(binary); - command["message_b64"] = base64_encode(message); - return(command); -} - -DValue websocket_exec_make_close_command(String connection_id, u16 status_code = 1000, String reason = "") -{ - DValue command; - command["action"] = "close"; - command["connection_id"] = connection_id; - command["status_code"] = (f64)status_code; - command["reason"] = reason; - return(command); -} - -void websocket_exec_apply_command(DValue command) -{ - String action = command["action"].to_string(); - if(action == "broadcast") - { - bool ok = false; - String payload = base64_decode(command["message_b64"].to_string(), ok); - if(!ok) - return; - server.websocket_broadcast(command["scope"].to_string(), payload, command["binary"].to_bool()); - return; - } - if(action == "send_to") - { - bool ok = false; - String payload = base64_decode(command["message_b64"].to_string(), ok); - if(!ok) - return; - server.websocket_send_to(command["connection_id"].to_string(), payload, command["binary"].to_bool()); - return; - } - if(action == "close") - { - server.websocket_close( - command["connection_id"].to_string(), - (u16)command["status_code"].to_u64(), - command["reason"].to_string() - ); - } -} - -void websocket_exec_apply_result(DValue result) -{ - if(result["type"].to_string() != "result") - return; - - String inflight_connection_id = websocket_exec_inflight_job["connection_id"].to_string(); - String result_connection_id = result["connection_id"].to_string(); - if(inflight_connection_id != "" && result_connection_id != "" && inflight_connection_id != result_connection_id) - { - printf("(!) websocket dispatch result mismatch: expected %s got %s\n", - inflight_connection_id.c_str(), - result_connection_id.c_str()); - } - - FastCGIServer::Connection* connection = websocket_find_connection(result_connection_id); - if(connection) - connection->websocket_state = result["connection_state"]; - - result["commands"].each([] (DValue command, String) { - websocket_exec_apply_command(command); - }); - - websocket_exec_inflight_job.clear(); -} - -void websocket_exec_handle_ipc_line(String line) -{ - line = trim(line); - if(line == "") - return; - DValue result = json_decode(line); - websocket_exec_apply_result(result); -} - -void websocket_exec_read_results() -{ - if(websocket_exec_fd == -1) - return; - - char buffer[4096]; - for(;;) - { - ssize_t read_result = read(websocket_exec_fd, buffer, sizeof(buffer)); - if(read_result == 0) - { - printf("(!) websocket executor disconnected\n"); - websocket_exec_fail_inflight_job(); - websocket_exec_clear_ipc_state(); - return; - } - if(read_result < 0) - { - if(errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR) - break; - perror("websocket executor read"); - websocket_exec_fail_inflight_job(); - websocket_exec_clear_ipc_state(); - return; - } - - websocket_exec_read_buffer.append(buffer, read_result); - for(;;) - { - size_t line_end = websocket_exec_read_buffer.find('\n'); - if(line_end == String::npos) - break; - String line = websocket_exec_read_buffer.substr(0, line_end); - websocket_exec_read_buffer.erase(0, line_end + 1); - websocket_exec_handle_ipc_line(line); - } - } -} - -void websocket_exec_flush_queue() -{ - if(websocket_exec_fd == -1) - return; - - if(websocket_exec_write_buffer == "" && !websocket_exec_has_inflight_job() && websocket_exec_pending_jobs.size() > 0) - { - websocket_exec_inflight_job = websocket_exec_pending_jobs.front(); - websocket_exec_pending_jobs.pop_front(); - websocket_exec_write_buffer = websocket_exec_inflight_job["serialized"].to_string(); - } - - while(websocket_exec_write_buffer != "") - { - ssize_t write_result = write(websocket_exec_fd, websocket_exec_write_buffer.data(), websocket_exec_write_buffer.length()); - if(write_result < 0) - { - if(errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR) - return; - perror("websocket executor write"); - websocket_exec_fail_inflight_job(); - websocket_exec_clear_ipc_state(); - return; - } - if(write_result == 0) - return; - websocket_exec_write_buffer.erase(0, write_result); - } -} - -Request websocket_exec_build_event_request(DValue job, String message) -{ - Request event_request; - event_request.server = &server_state; - event_request.params = job["params"].to_stringmap(); - event_request.params["REQUEST_METHOD"] = "WEBSOCKET"; - prepare_request_body_maps(event_request); - event_request.resources.is_websocket = true; - event_request.resources.websocket_connection_id = job["connection_id"].to_string(); - event_request.resources.websocket_scope = job["scope"].to_string(); - event_request.resources.websocket_opcode = (u8)job["opcode"].to_u64(); - event_request.resources.websocket_is_binary = job["is_binary"].to_bool(); - event_request.resources.websocket_is_text = job["is_text"].to_bool(); - event_request.resources.websocket_dispatch_capture = true; - job["scope_connections"].each([&] (DValue item, String) { - event_request.resources.websocket_scope_connection_ids.push_back(item.to_string()); - }); - event_request.connection = job["connection_state"]; - event_request.stats.time_init = time_precise(); - event_request.stats.time_start = event_request.stats.time_init; - event_request.random_index = 0; - event_request.random_seed = request_seed_from_time(event_request.stats.time_start); - event_request.response_code = "WEBSOCKET"; - event_request.header["Content-Type"] = server_state.config["CONTENT_TYPE"]; - event_request.in = message; - - event_request.params["WS_MESSAGE"] = message; - event_request.params["WS_CONNECTION_ID"] = event_request.resources.websocket_connection_id; - event_request.params["WS_SCOPE"] = event_request.resources.websocket_scope; - event_request.params["WS_CONNECTION_COUNT"] = (f64)event_request.resources.websocket_scope_connection_ids.size(); - event_request.params["WS_OPCODE"] = (f64)event_request.resources.websocket_opcode; - event_request.params["WS_MESSAGE_TYPE"] = (event_request.resources.websocket_is_binary ? "BINARY" : "TEXT"); - event_request.params["WS_DOCUMENT_URI"] = first( - event_request.params["DOCUMENT_URI"], - event_request.params["REQUEST_URI"] - ); - - return(event_request); -} - -bool websocket_exec_send_response(int fd, DValue response) -{ - String encoded = json_encode(response) + "\n"; - size_t offset = 0; - while(offset < encoded.length()) - { - ssize_t write_result = write(fd, encoded.data() + offset, encoded.length() - offset); - if(write_result < 0) - { - if(errno == EINTR) - continue; - return(false); - } - offset += (size_t)write_result; - } - return(true); -} - -void websocket_exec_process_job_line(int fd, String line) -{ - line = trim(line); - if(line == "") - return; - - DValue job = json_decode(line); - if(job["type"].to_string() != "dispatch") - return; - - bool decoded = false; - String message = base64_decode(job["message_b64"].to_string(), decoded); - if(!decoded) - { - printf("(!) invalid websocket IPC payload for %s\n", job["connection_id"].to_string().c_str()); - return; - } - - Request event_request = websocket_exec_build_event_request(job, message); - Request* previous_context = set_active_request(event_request); - server_state.request_count += 1; - - compiler_invoke_websocket(&event_request, event_request.params["SCRIPT_FILENAME"]); - - if(event_request.session_id.length() > 0) - save_session_data(event_request.session_id, event_request.session); - cleanup_mysql_connections(); - cleanup_sqlite_connections(); - - DValue response; - response["type"] = "result"; - response["connection_id"] = event_request.resources.websocket_connection_id; - response["connection_state"] = event_request.connection; - response["commands"] = event_request.resources.websocket_dispatch_commands; - - restore_active_request(previous_context); - if(!websocket_exec_send_response(fd, response)) - exit(1); -} - -void websocket_exec_child_loop(int fd) -{ - Request background_context; - my_pid = getpid(); - context = &background_context; - close_inherited_server_sockets(); - signal(SIGSEGV, on_segfault); - signal(SIGABRT, on_segfault); - signal(SIGBUS, on_segfault); - signal(SIGILL, on_segfault); - signal(SIGFPE, on_segfault); - signal(SIGPIPE, SIG_IGN); - setpriority(PRIO_PROCESS, 0, 5); - - String read_buffer; - char buffer[4096]; - for(;;) - { - ssize_t read_result = read(fd, buffer, sizeof(buffer)); - if(read_result == 0) - exit(0); - if(read_result < 0) - { - if(errno == EINTR) - continue; - exit(1); - } - - read_buffer.append(buffer, read_result); - for(;;) - { - size_t line_end = read_buffer.find('\n'); - if(line_end == String::npos) - break; - String line = read_buffer.substr(0, line_end); - read_buffer.erase(0, line_end + 1); - websocket_exec_process_job_line(fd, line); - } - } -} - -bool websocket_exec_alive() -{ - return(websocket_exec_pid > 0 && task_kill(websocket_exec_pid, 0) == 0 && websocket_exec_fd != -1); -} - -void ensure_websocket_executor() -{ - if(!websocket_exec_enabled_for_process()) - return; - if(websocket_exec_alive()) - return; - - if(websocket_exec_pid > 0 || websocket_exec_fd != -1) - { - websocket_exec_fail_inflight_job(); - websocket_exec_clear_ipc_state(); - websocket_exec_pid = 0; - } - - int sockets[2] = {-1, -1}; - if(socketpair(AF_UNIX, SOCK_STREAM, 0, sockets) != 0) - { - perror("socketpair"); - return; - } - - pid_t p = fork(); - if(p < 0) - { - perror("fork"); - close(sockets[0]); - close(sockets[1]); - return; - } - if(p == 0) - { - parent_pid = getppid(); - file_release_process_locks("websocket executor fork"); - prctl(PR_SET_PDEATHSIG, SIGHUP); - close(sockets[0]); - websocket_exec_child_loop(sockets[1]); - exit(0); - } - - close(sockets[1]); - if(!websocket_ipc_set_nonblocking(sockets[0])) - { - printf("(!) failed to set websocket executor socket nonblocking\n"); - close(sockets[0]); - return; - } - - websocket_exec_fd = sockets[0]; - websocket_exec_pid = p; - printf("(P) websocket executor spawned: PID %i\n", p); -} - -void websocket_exec_tick() -{ - if(!websocket_exec_enabled_for_process()) - return; - if(!websocket_exec_alive()) - { - websocket_exec_fail_inflight_job(); - websocket_exec_clear_ipc_state(); - websocket_exec_pid = 0; - ensure_websocket_executor(); - } - websocket_exec_read_results(); - websocket_exec_flush_queue(); -} - -} String current_ws_scope() { @@ -682,13 +218,6 @@ bool ws_is_binary() StringList ws_connections(String scope) { - if(context && context->resources.websocket_dispatch_capture) - { - String normalized_scope = normalize_ws_scope(scope); - if(normalized_scope == context->resources.websocket_scope) - return(context->resources.websocket_scope_connection_ids); - return(StringList()); - } return(server.websocket_connection_ids(normalize_ws_scope(scope))); } @@ -699,26 +228,11 @@ u64 ws_connection_count(String scope) bool ws_send(String message, bool binary, String scope) { - String normalized_scope = normalize_ws_scope(scope); - if(context && context->resources.websocket_dispatch_capture) - { - DValue command = websocket_exec_make_message_command("broadcast", message, binary); - command["scope"] = normalized_scope; - websocket_exec_append_command(command); - return(true); - } - return(server.websocket_broadcast(normalized_scope, message, binary) > 0); + return(server.websocket_broadcast(normalize_ws_scope(scope), message, binary) > 0); } bool ws_send_to(String connection_id, String message, bool binary) { - if(context && context->resources.websocket_dispatch_capture) - { - DValue command = websocket_exec_make_message_command("send_to", message, binary); - command["connection_id"] = connection_id; - websocket_exec_append_command(command); - return(true); - } return(server.websocket_send_to(connection_id, message, binary)); } @@ -728,11 +242,6 @@ bool ws_close(String connection_id) connection_id = ws_connection_id(); if(connection_id == "") return(false); - if(context && context->resources.websocket_dispatch_capture) - { - websocket_exec_append_command(websocket_exec_make_close_command(connection_id)); - return(true); - } return(server.websocket_close(connection_id)); } @@ -861,7 +370,7 @@ int handle_cli_complete(FastCGIRequest& request) String cli_unit = compiler_normalize_unit_path(&request, script_filename); if(wasm_backend_should_handle(request, cli_unit)) { - String wasm_error = wasm_backend_serve(request, cli_unit, wasm_kind::CLI); + String wasm_error = wasm_backend_serve(request, cli_unit, "cli"); if(wasm_error != "") { request.set_status(500, "Internal Server Error"); @@ -913,11 +422,6 @@ int handle_data(FastCGIRequest& request) { } int handle_complete(FastCGIRequest& request) { - // The event handler can also be a class member function. This - // event occurs when the parameters and standard input streams are - // both closed, and thus the request is complete. - // printf("(i) request handle\n"); - Request* previous_context = set_active_request(request); server_state.request_count += 1; request.server = &server_state; @@ -958,10 +462,10 @@ int handle_complete(FastCGIRequest& request) { // wasm invocation: Wasmtime uses host signals internally to implement // guest traps, and the native handler would otherwise turn a clean // guest trap into a native fatal signal. Sets failure_* on error. - auto serve_via_wasm = [&](const String& entry_unit, int32_t kind) { + auto serve_via_wasm = [&](const String& entry_unit, const String& handler) { request_fault_active = 0; restore_request_fault_handlers(); - String wasm_error = wasm_backend_serve(request, entry_unit, kind); + String wasm_error = wasm_backend_serve(request, entry_unit, handler); install_request_fault_handlers(); request_fault_active = 1; if(wasm_error != "") @@ -973,22 +477,54 @@ int handle_complete(FastCGIRequest& request) { }; String entry_unit = compiler_normalize_unit_path(&request, request.params["SCRIPT_FILENAME"]); - if(request.resources.is_cli) + if(request.params["UCE_WS"] == "1") + { + // A WS message the broker forwarded here: rebuild the connection + // context the broker passed as params, then run __uce_websocket. + request.resources.websocket_connection_id = request.params["UCE_WS_CONNECTION_ID"]; + request.resources.websocket_scope = request.params["UCE_WS_SCOPE"]; + request.resources.websocket_opcode = (u8)int_val(request.params["UCE_WS_OPCODE"]); + request.resources.websocket_is_binary = request.params["UCE_WS_BINARY"] == "1"; + bool msg_ok = false; + request.in = base64_decode(request.params["UCE_WS_MESSAGE"], msg_ok); + for(auto& id : split(request.params["UCE_WS_CONNECTIONS"], "\n")) + if(id != "") + request.resources.websocket_scope_connection_ids.push_back(id); + bool decoded = false; + String state_raw = base64_decode(request.params["UCE_WS_STATE"], decoded); + if(state_raw != "") + { + DValue state; String e; + if(ucb_decode(state_raw, state, &e)) + request.connection = state; + } + if(wasm_backend_should_handle(request, entry_unit)) + serve_via_wasm(entry_unit, "websocket"); + else + compiler_invoke_websocket(&request, request.params["SCRIPT_FILENAME"]); + } + else if(request.resources.is_cli) { if(wasm_backend_should_handle(request, entry_unit)) - serve_via_wasm(entry_unit, wasm_kind::CLI); + serve_via_wasm(entry_unit, "cli"); else compiler_invoke_cli(&request, request.params["SCRIPT_FILENAME"]); } else if(request.params["UCE_SERVE_HTTP"] == "1") - // W7c pending: the custom-server dispatcher is forked from a worker - // that already holds a live Wasmtime engine, so re-creating an engine - // in that child currently hangs under the in-process dispatcher path. - // The fix is to have the broker forward to the worker pool rather than - // render in the fork; until then serve_http renders natively. - compiler_invoke_serve_http(&request, request.params["SCRIPT_FILENAME"], request.params["UCE_SERVE_HTTP_FUNCTION"]); + { + // W7c: this runs in a normal worker (clean engine) — the custom-server + // dispatcher forwarded the request here via FastCGI rather than render + // wasm in its own fork (Wasmtime cannot be re-created across fork). + if(wasm_backend_should_handle(request, entry_unit)) + { + String fn = request.params["UCE_SERVE_HTTP_FUNCTION"]; + serve_via_wasm(entry_unit, fn == "" ? String("serve_http") : "serve_http:" + fn); + } + else + compiler_invoke_serve_http(&request, request.params["SCRIPT_FILENAME"], request.params["UCE_SERVE_HTTP_FUNCTION"]); + } else if(wasm_backend_should_handle(request, entry_unit)) - serve_via_wasm(entry_unit, wasm_kind::RENDER); + serve_via_wasm(entry_unit, "render"); else compiler_invoke(&request, request.params["SCRIPT_FILENAME"]); } @@ -1027,40 +563,6 @@ int handle_complete(FastCGIRequest& request) { return request.flags.status; } -int handle_websocket_message(FastCGIRequest& request, const String& message, u8 opcode) -{ - ensure_websocket_executor(); - if(!websocket_exec_alive()) - { - printf("(!) websocket executor unavailable for %s\n", request.resources.websocket_connection_id.c_str()); - server.websocket_close(request.resources.websocket_connection_id, 1011, "websocket handler unavailable"); - return(0); - } - - DValue job; - job["type"] = "dispatch"; - job["connection_id"] = request.resources.websocket_connection_id; - job["scope"] = request.resources.websocket_scope; - job["opcode"] = (f64)opcode; - job["is_binary"].set_bool(request.resources.websocket_is_binary); - job["is_text"].set_bool(request.resources.websocket_is_text); - job["message_b64"] = base64_encode(message); - if(request.resources.websocket_connection_state) - job["connection_state"] = *request.resources.websocket_connection_state; - - for(auto& item : request.params) - job["params"][item.first] = item.second; - - for(auto& connection_id : websocket_exec_snapshot_connections(request.resources.websocket_scope)) - { - DValue snapshot_item; - snapshot_item = connection_id; - job["scope_connections"].push(snapshot_item); - } - - websocket_exec_queue_job(job); - return 0; -} volatile bool termination_signal_received = false; @@ -1218,6 +720,31 @@ StringMap custom_server_read_config(String key) static String custom_server_dispatcher_key = ""; +// Forward a request to a normal worker over the FastCGI socket and populate the +// response from the worker's reply. Shared by the connection brokers (the custom +// HTTP server dispatcher and the WS broker): they own a listening socket but +// render through the clean-engine worker pool rather than host a Wasmtime engine +// (which cannot be re-created across fork). The caller sets the routing params +// (UCE_SERVE_HTTP / UCE_WS, SCRIPT_FILENAME, etc.) before calling. +int forward_request_to_worker(FastCGIRequest& request, u32 timeout_seconds = 30) +{ + String fcgi_socket = first(server_state.config["FCGI_SOCKET_PATH"], "/run/uce.sock"); + FcgiForwardResult fwd = fcgi_forward_request(fcgi_socket, request.params, request.in, timeout_seconds); + request.ob_start(); + if(!fwd.ok) + { + request.set_status(502, "Bad Gateway"); + request.header["Content-Type"] = "text/plain; charset=utf-8"; + (*request.ob) << "worker forward failed: " << fwd.error << "\n"; + return(request.flags.status); + } + request.set_status(fwd.status); + for(auto& h : fwd.headers) + request.header[h.first] = h.second; + request.ob->write(fwd.body.data(), fwd.body.size()); + return(request.flags.status); +} + int custom_server_http_complete(FastCGIRequest& request) { String key = first(request.params["UCE_SERVER_KEY"], custom_server_dispatcher_key); @@ -1237,12 +764,193 @@ int custom_server_http_complete(FastCGIRequest& request) request.params["UCE_SERVE_HTTP_FUNCTION"] = cfg["function"]; request.params["SCRIPT_FILENAME"] = cfg["file"]; u64 timeout = config_map_u64(server_state.config, "CUSTOM_SERVER_HANDLER_TIMEOUT_SECONDS", 30); - if(timeout > 0) - alarm(timeout); - int status = handle_complete(request); - if(timeout > 0) - alarm(0); - return(status); + return(forward_request_to_worker(request, timeout > 0 ? (u32)timeout : 30)); +} + +// ---- central WS broker ----------------------------------------------------- + +// Connect to a unix socket (blocking — local, sub-ms) and set it non-blocking. +static int ws_broker_connect_unix(const String& path) +{ + int fd = ::socket(AF_UNIX, SOCK_STREAM, 0); + if(fd < 0) + return(-1); + struct sockaddr_un addr; + memset(&addr, 0, sizeof(addr)); + addr.sun_family = AF_UNIX; + strncpy(addr.sun_path, path.c_str(), sizeof(addr.sun_path) - 1); + if(::connect(fd, (struct sockaddr*)&addr, sizeof(addr)) < 0) + { + ::close(fd); + return(-1); + } + int fl = fcntl(fd, F_GETFL, 0); + fcntl(fd, F_SETFL, fl | O_NONBLOCK); + return(fd); +} + +// Apply a ws_* dispatch batch a worker flushed at workspace teardown. The broker +// owns every connection, so any target (id / scope / broadcast / close) resolves. +void ws_broker_apply_commands(FastCGIRequest& request) +{ + DValue batch; + String err; + bool ok = ucb_decode(request.in, batch, &err); + if(ok) + { + if(DValue* cmds = batch.key("commands")) + cmds->each([&](const DValue& cmd_const, String) { + DValue cmd = cmd_const; // .each yields const; copy for [] access + String action = cmd["action"].to_string(); + bool ok = false; + String msg = base64_decode(cmd["message_b64"].to_string(), ok); + bool binary = cmd["binary"].to_bool(); + if(action == "broadcast") + ws_broker.websocket_broadcast(cmd["scope"].to_string(), msg, binary); + else if(action == "send_to") + ws_broker.websocket_send_to(cmd["connection_id"].to_string(), msg, binary); + else if(action == "close") + ws_broker.websocket_close(cmd["connection_id"].to_string(), + (u16)cmd["status_code"].to_u64(), cmd["reason"].to_string()); + }); + // persist any updated per-connection state back onto the live connection + String cid = batch["connection_id"].to_string(); + if(cid != "" && batch.key("connection_state")) + for(auto& item : ws_broker.client_sockets) + if(item.second->is_websocket && item.second->websocket_connection_id == cid) + item.second->websocket_state = batch["connection_state"]; + } + request.set_status(200); + request.ob_start(); +} + +// on_complete for the broker: either a worker's command flush, or an un-upgraded +// HTTP request that hit the WS port (forwarded to the pool via the shared facility). +int ws_broker_complete(FastCGIRequest& request) +{ + if(request.params["UCE_WS_DISPATCH"] == "1") + { + ws_broker_apply_commands(request); + return(request.flags.status); + } + return(forward_request_to_worker(request)); +} + +// A complete WS message: fire a render to the worker pool WITHOUT blocking the +// broker loop (the connection identity + state ride as params; the message is +// the body). The worker renders __uce_websocket and flushes ws_* back to us. +int ws_broker_ws_message(FastCGIRequest& request, const String& message, u8 opcode) +{ + StringMap params; + params["SCRIPT_FILENAME"] = request.params["SCRIPT_FILENAME"]; + params["REQUEST_METHOD"] = "GET"; + // handle_request() rejects any request without REQUEST_URI before on_complete. + params["REQUEST_URI"] = first(request.params["REQUEST_URI"], request.params["DOCUMENT_URI"], + request.params["SCRIPT_FILENAME"]); + params["UCE_WS"] = "1"; + // The message rides as a param (empty STDIN) so the forwarded request + // completes cleanly — a STDIN body makes the FastCGI transport flush a + // premature response before on_complete (handle_complete) ever runs. + params["UCE_WS_MESSAGE"] = base64_encode(message); + params["UCE_WS_CONNECTION_ID"] = request.resources.websocket_connection_id; + params["UCE_WS_SCOPE"] = request.resources.websocket_scope; + params["UCE_WS_OPCODE"] = std::to_string((int)opcode); + params["UCE_WS_BINARY"] = request.resources.websocket_is_binary ? "1" : "0"; + params["UCE_WS_CONNECTIONS"] = join(ws_broker.websocket_connection_ids(request.resources.websocket_scope), "\n"); + if(request.resources.websocket_connection_state) + params["UCE_WS_STATE"] = base64_encode(ucb_encode(*request.resources.websocket_connection_state)); + + int fd = ws_broker_connect_unix(first(server_state.config["FCGI_SOCKET_PATH"], "/run/uce.sock")); + if(fd < 0) + return(0); + ws_broker_outbound[fd] = fcgi_build_request(params, ""); + return(0); +} + +// Drive in-flight render forwards non-blocking: finish writing the request, then +// drain/discard the reply (the unit's ws_* came back via the command socket). +void ws_broker_drain_outbound() +{ + for(auto it = ws_broker_outbound.begin(); it != ws_broker_outbound.end(); ) + { + int fd = it->first; + String& pending = it->second; + bool done = false; + if(!pending.empty()) + { + ssize_t n = ::send(fd, pending.data(), pending.size(), MSG_NOSIGNAL | MSG_DONTWAIT); + if(n > 0) + pending.erase(0, n); + else if(n < 0 && errno != EAGAIN && errno != EWOULDBLOCK) + done = true; + } + if(!done && pending.empty()) + { + char buf[8192]; + ssize_t r = ::recv(fd, buf, sizeof(buf), MSG_DONTWAIT); + if(r == 0) + done = true; // worker closed the connection: render complete + else if(r < 0 && errno != EAGAIN && errno != EWOULDBLOCK) + done = true; + } + if(done) + { + ::close(fd); + it = ws_broker_outbound.erase(it); + } + else + ++it; + } +} + +void run_ws_broker() +{ + my_pid = getpid(); + close_inherited_server_sockets(); // drop the worker listeners we inherited + install_process_fault_handlers(); + ws_broker.calls_until_termination = -1; + // The broker renders nothing itself, so accept every request through to + // on_complete (it either forwards to the pool or applies a ws_* batch). + ws_broker.on_request = [](FastCGIRequest&) { return 0; }; + ws_broker.on_data = [](FastCGIRequest&) { return 0; }; + ws_broker.on_complete = &ws_broker_complete; + ws_broker.on_websocket_message = &ws_broker_ws_message; + if(server_state.config["HTTP_PORT"] != "") + ws_broker.listen_http(int_val(server_state.config["HTTP_PORT"])); + if(server_state.config["WS_BROKER_SOCKET_PATH"] != "") + { + ws_broker.listen(server_state.config["WS_BROKER_SOCKET_PATH"]); + chmod(server_state.config["WS_BROKER_SOCKET_PATH"].c_str(), S_IRWXU | S_IRGRP | S_IWGRP); + } + for(;;) + { + ws_broker.process(50); + ws_broker_drain_outbound(); + } +} + +bool ws_broker_alive() +{ + return(ws_broker_pid > 0 && task_kill(ws_broker_pid, 0) == 0); +} + +void ensure_ws_broker() +{ + if(ws_broker_alive()) + return; + pid_t p = fork(); + if(p < 0) + { + perror("fork ws_broker"); + return; + } + if(p == 0) + { + run_ws_broker(); + exit(0); + } + ws_broker_pid = p; + printf("(P) WS broker spawned: PID %i\n", p); } void custom_server_http_dispatcher_loop(String key) @@ -1512,29 +1220,17 @@ void ensure_proactive_compiler() void listen_for_connections() { install_process_fault_handlers(); - if(worker_accepts_http) - { - // Keep the dedicated HTTP/WebSocket worker alive. If it ages out like a - // normal FastCGI worker, nginx can connect to the shared listening socket - // while no child is actively accepting, which makes `.ws.uce` page loads - // appear to hang until the parent respawns a replacement worker. - server.calls_until_termination = -1; - } - if(!worker_accepts_http) - server.close_http_listeners(); + // Workers are uniform FastCGI/CLI renderers; the WS broker owns the HTTP/WS + // port and every connection, so workers never accept raw HTTP themselves. + server.close_http_listeners(); server.on_request = &handle_request; server.on_data = &handle_data; server.on_complete = &handle_complete; server.on_cli_complete = &handle_cli_complete; - server.on_websocket_message = &handle_websocket_message; - if(worker_accepts_http) - ensure_websocket_executor(); for(;;) { file_release_process_locks("worker loop cleanup"); - server.process(worker_accepts_http ? 50 : -1); - if(worker_accepts_http) - websocket_exec_tick(); + server.process(-1); } } @@ -1566,8 +1262,8 @@ void init_base_process() chmod(server_state.config["CLI_SOCKET_PATH"].c_str(), S_IRWXU | S_IRGRP | S_IWGRP); } - if(server_state.config["HTTP_PORT"] != "") - server.listen_http(int_val(server_state.config["HTTP_PORT"])); + // HTTP_PORT (WebSocket + raw HTTP) is owned by the dedicated WS broker, not + // the worker pool — see ensure_ws_broker(). Workers handle FastCGI + CLI only. mkdir(server_state.config["BIN_DIRECTORY"]); mkdir(server_state.config["TMP_UPLOAD_PATH"]); @@ -1588,6 +1284,7 @@ int main(int argc, char** argv) init_base_process(); ensure_proactive_compiler(); + ensure_ws_broker(); for(;;) { @@ -1596,15 +1293,18 @@ int main(int argc, char** argv) if(!termination_signal_received) ensure_proactive_compiler(); + // One dedicated WS broker owns all connections; respawn it if it dies + // (live connections are lost on a broker restart, but unit-code crashes + // happen in workers, not here, so the broker stays up in practice). + if(!ws_broker_alive()) + ws_broker_pid = 0; + if(!termination_signal_received) + ensure_ws_broker(); + while(workers.size() < int_val(server_state.config["WORKER_COUNT"])) { if(!termination_signal_received) - { - worker_accepts_http = (http_worker_pid == 0 || workers.count(http_worker_pid) == 0); - pid_t child_pid = spawn_subprocess(listen_for_connections); - if(child_pid > 0 && worker_accepts_http) - http_worker_pid = child_pid; - } + spawn_subprocess(listen_for_connections); } sleep(1); } diff --git a/src/wasm/backend.cpp b/src/wasm/backend.cpp index bdde4aa..b2bb2b3 100644 --- a/src/wasm/backend.cpp +++ b/src/wasm/backend.cpp @@ -19,6 +19,8 @@ #include #include #include +// Worker flushes ws_* dispatch batches to the broker via the FastCGI client. +#include "../lib/fcgi_forward.h" // per forked worker process: one engine + compiled-core cache, one epoch ticker static WasmWorker* g_wasm_worker = 0; @@ -116,18 +118,13 @@ static bool wasm_backend_native_fallback_uncached(Request* context, const String // Supported by the wasm core, intentionally NOT in this list: // - regex_* (host PCRE2 hostcall), xml_*/yaml_*/markdown_* (compiled in), // - unit_render()/component() (host resolver). - // What remains is genuinely host-owned / native-only for now: - // - zip pages that use C++ try/catch around zip_* error cases; - // - direct compiler_load_shared_unit() access to native SharedUnit internals. - // Background tasks, sleep/usleep, sockets/custom servers, memcache, mysql, - // and unit_call/unit_compile/unit_info/units_list are host-owned but now have - // membrane hostcalls, so they intentionally do not fallback here. - StringList native_only_tokens = { - "zip_", "compiler_load_shared_unit(" - }; - for(auto& token : native_only_tokens) - if(source.find(token) != String::npos) - return(true); + // W7d: every former native-only surface now goes through the membrane — + // zip_* (uce_host_zip), unit introspection (uce_host_units: unit_info/ + // unit_call/units_list), regex/xml/yaml/markdown, filesystem, sqlite, + // background tasks, sleep, sockets/custom servers, memcache, mysql. No unit + // source token forces native anymore, so there is nothing left to scan for. + // (The native fallback now only covers cold/stale artifacts; W7e removes it.) + (void)source; return(false); } @@ -168,11 +165,12 @@ bool wasm_backend_should_handle(Request& request, const String& entry_unit) } // Serve a request through a wasm workspace using the unit handler selected by -// `kind` (page render or cli). Populates the native Request -// (status/headers/cookies/session/body) so the existing transport writes the -// response unchanged. Returns "" on success, or a collapsed error/trace string -// for the caller to route into the configured error page. -String wasm_backend_serve(Request& request, const String& entry_unit, int32_t kind = WasmWorkspace::RESOLVE_RENDER) +// `kind` (page render / cli / serve_http, with an optional named serve_http +// handler). Populates the native Request (status/headers/cookies/session/body) +// so the existing transport writes the response unchanged. Returns "" on +// success, or a collapsed error/trace string for the caller to route into the +// configured error page. +String wasm_backend_serve(Request& request, const String& entry_unit, const String& handler = "render") { DValue ctx; auto copy_map = [&](const StringMap& source, const char* key) { @@ -186,19 +184,56 @@ String wasm_backend_serve(Request& request, const String& entry_unit, int32_t ki copy_map(request.session, "session"); ctx["entry_unit"] = entry_unit; // Raw request body: cli_input() parses a JSON CLI payload from context.in, - // carried into the workspace, not just the form-decoded post map. + // and serve_http handlers read it as req->in; carried into the workspace. ctx["in"] = request.in; + // WebSocket event context: the workspace owns no connections, so the frame's + // connection identity goes in and the handler's ws_send/ws_close dispatch + // commands come back out (below) for the native broker to apply. + if(handler == "websocket") + { + ctx["ws"]["connection_id"] = request.resources.websocket_connection_id; + ctx["ws"]["scope"] = request.resources.websocket_scope; + ctx["ws"]["opcode"] = (f64)request.resources.websocket_opcode; + ctx["ws"]["binary"].set_bool(request.resources.websocket_is_binary); + for(auto& id : request.resources.websocket_scope_connection_ids) + { + DValue v; v = id; + ctx["ws"]["connections"].push(v); + } + ctx["ws"]["connection_state"] = request.connection; + } - WasmResponse response = wasm_worker_serve(*g_wasm_worker, ctx, entry_unit, kind); + WasmResponse response = wasm_worker_serve(*g_wasm_worker, ctx, entry_unit, handler); if(!response.ok) return(response.error == "" ? String("wasm workspace failed") : response.error); - // A cli unit that does not export __uce_cli is a 404, matching native - // compiler_invoke_cli ("CLI Entry Point Not Found"). For page render a - // missing handler is simply an empty body (native parity). - if(!response.handler_present && kind == WasmWorkspace::RESOLVE_CLI) + // Any handler may have called ws_send/ws_close (not just WS handlers). If it + // left dispatch commands, flush the batch to the central WS broker, which owns + // all connections and resolves each command's target (id/scope/broadcast) + // against the full registry. This is the only path WS data takes out — the + // workspace owns no connections. + if(DValue* cmds = response.meta.key("ws_commands")) { - request.set_status(404, "CLI Entry Point Not Found"); + DValue batch; + batch["commands"] = *cmds; + if(DValue* cstate = response.meta.key("ws_connection_state")) + { + batch["connection_id"] = request.resources.websocket_connection_id; + batch["connection_state"] = *cstate; + } + StringMap dispatch_params; + dispatch_params["UCE_WS_DISPATCH"] = "1"; + String broker_socket = first(request.server ? request.server->config["WS_BROKER_SOCKET_PATH"] : String(), + "/run/uce/ws-broker.sock"); + fcgi_forward_request(broker_socket, dispatch_params, ucb_encode(batch), 5); + } + + // A cli/serve_http unit that does not export the requested handler is a 404, + // matching native compiler_invoke_cli. For page render a missing handler is + // simply an empty body (native parity). + if(!response.handler_present && handler != "render") + { + request.set_status(404, handler == "cli" ? "CLI Entry Point Not Found" : "Handler Not Found"); return(""); } diff --git a/src/wasm/backend.h b/src/wasm/backend.h index 5f182e4..0c9e516 100644 --- a/src/wasm/backend.h +++ b/src/wasm/backend.h @@ -6,27 +6,15 @@ // backend.cpp — so it does not have to compile worker.cpp + wasmtime.hh on // every build. Include only after uce_lib.h (needs Request / String). -#include - -// Resolve-kind selector for wasm_backend_serve, mirrored from -// WasmWorkspace::ResolveKind in worker.cpp. Only the entry kinds actually wired -// to the wasm backend live here; WebSocket/serve_http re-enter when W7b/W7c land -// (via the broker→worker-pool forwarding model, not in-fork rendering). -namespace wasm_kind { - enum { - RENDER = 1, - CLI = 4, - }; -} - // True if this request should be served by the wasm backend (config + artifact -// + fallback-token gate); mode-agnostic, the caller picks the kind. +// + fallback-token gate); handler-agnostic, the caller names the handler. bool wasm_backend_should_handle(Request& request, const String& entry_unit); -// Serve a request through a wasm workspace using the unit handler for `kind`, -// populating the native Request. Returns "" on success or a collapsed error. -String wasm_backend_serve(Request& request, const String& entry_unit, - int32_t kind = wasm_kind::RENDER); +// Serve a request through a wasm workspace by invoking a named unit handler — +// "render", "cli", "websocket", "serve_http", "serve_http:named" — and populate +// the native Request. Returns "" on success or a collapsed error. The handler is +// just an export name; there is no per-mode machinery. +String wasm_backend_serve(Request& request, const String& entry_unit, const String& handler = "render"); // Join the per-process epoch ticker before the worker process exits. void wasm_backend_shutdown(); diff --git a/src/wasm/core.cpp b/src/wasm/core.cpp index 4574471..488b5bd 100644 --- a/src/wasm/core.cpp +++ b/src/wasm/core.cpp @@ -475,7 +475,8 @@ extern "C" void uce_wasm_link_anchors() // slots (loading units lazily) and writes the resolved unit path back so // nested relative component resolution keeps working. extern "C" int32_t uce_host_component_resolve( - const char* target, size_t target_len, int32_t kind, + const char* target, size_t target_len, + const char* handler, size_t handler_len, const char* current_unit, size_t current_unit_len, char* resolved_buf, size_t resolved_cap); @@ -534,38 +535,37 @@ struct RequestPropsScope } }; -// kind values shared with the host loader (src/wasm/worker.cpp) -enum WasmResolveKind { - WASM_RESOLVE_COMPONENT = 0, - WASM_RESOLVE_RENDER = 1, - WASM_RESOLVE_EXISTS = 2, - WASM_RESOLVE_ONCE = 3, - WASM_RESOLVE_CLI = 4, -}; - -static s32 wasm_resolve_target(String target, s32 kind, String* resolved_out = 0) +// A unit is a bag of exported handlers; invoking any of them is one operation — +// the host resolves __uce_ in the loaded module to a funcref slot. The +// handler is just a string: "render", "component:CARD", "render:VARIANT", +// "once", "cli", "websocket", "serve_http:named" — or "exists" (an existence +// probe that loads nothing). No per-mode kinds. +static s32 wasm_resolve_target(String unit_target, String handler, String* resolved_out = 0) { - String cache_key = std::to_string(kind) + ":" + target; + String cache_key = handler + "\t" + unit_target; + bool is_exists = (handler == "exists"); auto cached = wasm_component_slots.find(cache_key); - if(cached != wasm_component_slots.end() && kind != WASM_RESOLVE_EXISTS) + if(cached != wasm_component_slots.end() && !is_exists) return(cached->second); char resolved[512]; String current = context ? context->resources.current_unit_file : ""; s32 slot = uce_host_component_resolve( - target.data(), target.size(), kind, + unit_target.data(), unit_target.size(), handler.data(), handler.size(), current.data(), current.size(), resolved, sizeof(resolved)); if(resolved_out && slot) *resolved_out = String(resolved, strnlen(resolved, sizeof(resolved))); - if(kind != WASM_RESOLVE_EXISTS) + if(!is_exists) wasm_component_slots[cache_key] = slot; return(slot); } String component_resolve(String name) { + String file_name, render_name; + component_parse_target(trim(name), file_name, render_name); String resolved; - if(wasm_resolve_target(trim(name), WASM_RESOLVE_EXISTS, &resolved)) + if(wasm_resolve_target(file_name, "exists", &resolved)) return(resolved); return(""); } @@ -585,7 +585,7 @@ static void wasm_run_once(const String& resolved, Request& request) if(request.once_units.find(resolved) != request.once_units.end()) return; request.once_units.insert(resolved); - s32 once_slot = wasm_resolve_target(resolved, WASM_RESOLVE_ONCE); + s32 once_slot = wasm_resolve_target(resolved, "once"); if(once_slot == 0) return; String previous_unit = request.resources.current_unit_file; @@ -597,8 +597,11 @@ static void wasm_run_once(const String& resolved, Request& request) void component_render(String name, DValue props, Request& request) { + String file_name, render_name; + component_parse_target(trim(name), file_name, render_name); + String handler = render_name == "" ? String("component") : "component:" + render_name; String resolved; - s32 slot = wasm_resolve_target(trim(name), WASM_RESOLVE_COMPONENT, &resolved); + s32 slot = wasm_resolve_target(file_name, handler, &resolved); if(!slot) { print(component_error_banner("component not found: " + trim(name))); @@ -611,8 +614,8 @@ void component_render(String name, DValue props, Request& request) request.resources.current_unit_file = resolved; // a wasm function pointer is its index in the shared funcref table; the // host returned the handler's slot, so this is a plain call_indirect - request_ref_handler handler = (request_ref_handler)(uintptr_t)slot; - handler(request); + request_ref_handler handler_fn = (request_ref_handler)(uintptr_t)slot; + handler_fn(request); request.resources.current_unit_file = previous_unit; } @@ -633,8 +636,11 @@ String component(String name, DValue props) { return(component(name, props, *con void unit_render(String file_name, Request& request) { + String unit_name, render_name; + component_parse_target(trim(file_name), unit_name, render_name); + String handler = render_name == "" ? String("render") : "render:" + render_name; String resolved; - s32 slot = wasm_resolve_target(trim(file_name), WASM_RESOLVE_RENDER, &resolved); + s32 slot = wasm_resolve_target(unit_name, handler, &resolved); if(!slot) { print(component_error_banner("unit not found: " + trim(file_name))); @@ -644,8 +650,8 @@ void unit_render(String file_name, Request& request) String previous_unit = request.resources.current_unit_file; if(resolved != "") request.resources.current_unit_file = resolved; - request_ref_handler handler = (request_ref_handler)(uintptr_t)slot; - handler(request); + request_ref_handler handler_fn = (request_ref_handler)(uintptr_t)slot; + handler_fn(request); request.resources.current_unit_file = previous_unit; } @@ -653,25 +659,25 @@ void unit_render(String file_name) { unit_render(file_name, *context); } extern "C" { -// Host calls this to invoke the request's entry unit through one of its -// directive handlers (render/cli/websocket/serve_http). Routing through the -// resolver gives the entry the same ONCE() + dispatch semantics as components -// (the host pre-loaded it, so resolution is a cache hit). The host pre-checks -// that the handler export exists, so a missing slot here is a silent no-op. -void uce_wasm_invoke_entry(const char* path, size_t len, int32_t kind) +// Host calls this to invoke the request's entry unit through a named handler +// ("render"/"cli"/"websocket"/"serve_http:named"). It is the same resolve + +// ONCE() + dispatch path as components — the entry handler is just another +// export. The host pre-checks the export exists, so a missing slot is a no-op. +void uce_wasm_invoke_entry(const char* path, size_t path_len, const char* handler, size_t handler_len) { // the host always runs uce_wasm_core_init + apply_context before this - String file_name(path, len); + String file_name(path, path_len); + String handler_name(handler, handler_len); String resolved; - s32 slot = wasm_resolve_target(trim(file_name), kind, &resolved); + s32 slot = wasm_resolve_target(trim(file_name), handler_name, &resolved); if(!slot) return; wasm_run_once(resolved, *context); String previous_unit = context->resources.current_unit_file; if(resolved != "") context->resources.current_unit_file = resolved; - request_ref_handler handler = (request_ref_handler)(uintptr_t)slot; - handler(*context); + request_ref_handler handler_fn = (request_ref_handler)(uintptr_t)slot; + handler_fn(*context); context->resources.current_unit_file = previous_unit; } @@ -761,6 +767,27 @@ int uce_wasm_apply_context(const char* buf, size_t len) wasm_request.resources.current_unit_file = entry->to_string(); DValue* raw_in = decoded.key("in"); wasm_request.in = raw_in ? raw_in->to_string() : ""; + // websocket event context: ws_send()/ws_close() capture into the dispatch + // list (the workspace owns no connections), which collect() carries back to + // the broker. Reset per invocation. + wasm_request.resources.websocket_dispatch_commands = DValue(); + wasm_request.resources.websocket_dispatch_capture = false; + DValue* ws = decoded.key("ws"); + if(ws) + { + wasm_request.resources.websocket_connection_id = (*ws)["connection_id"].to_string(); + wasm_request.resources.websocket_scope = (*ws)["scope"].to_string(); + wasm_request.resources.websocket_opcode = (u8)(*ws)["opcode"].to_u64(); + wasm_request.resources.websocket_is_binary = (*ws)["binary"].to_bool(); + wasm_request.resources.websocket_scope_connection_ids.clear(); + if(DValue* conns = ws->key("connections")) + conns->each([&](const DValue& v, String) { + wasm_request.resources.websocket_scope_connection_ids.push_back(v.to_string()); + }); + wasm_request.resources.websocket_dispatch_capture = true; + if(DValue* cstate = ws->key("connection_state")) + wasm_request.connection = *cstate; + } return(0); } @@ -787,6 +814,14 @@ void uce_wasm_finish_response_meta() } for(auto& entry : wasm_request.session) meta["session"][entry.first] = entry.second; + // Any unit code (not just WS handlers) may call ws_send/ws_close; whenever the + // dispatch list is non-empty, carry it back so the worker can flush it to the + // broker. ws_connection_state rides along for stateful WS handlers. + if(!wasm_request.resources.websocket_dispatch_commands._map.empty()) + { + meta["ws_commands"] = wasm_request.resources.websocket_dispatch_commands; + meta["ws_connection_state"] = wasm_request.connection; + } wasm_response_meta = ucb_encode(meta); } diff --git a/src/wasm/core_hostcalls.syms b/src/wasm/core_hostcalls.syms index 947337d..ae9c68b 100644 --- a/src/wasm/core_hostcalls.syms +++ b/src/wasm/core_hostcalls.syms @@ -22,5 +22,8 @@ uce_host_file_exists uce_host_file_read uce_host_file_write uce_host_file_unlink +uce_host_file_list +uce_host_file_mkdir +uce_host_file_mtime uce_host_regex uce_host_sqlite diff --git a/src/wasm/worker.cpp b/src/wasm/worker.cpp index dfd04e7..dfb4b92 100644 --- a/src/wasm/worker.cpp +++ b/src/wasm/worker.cpp @@ -42,6 +42,9 @@ #include #include #include +#include +#include +#include struct WasmDylinkInfo { @@ -388,8 +391,6 @@ public: // resolve-kind values shared with the guest core (src/wasm/core.cpp) // must match WasmResolveKind in src/wasm/core.cpp - enum ResolveKind { RESOLVE_COMPONENT = 0, RESOLVE_RENDER = 1, RESOLVE_EXISTS = 2, RESOLVE_ONCE = 3, - RESOLVE_CLI = 4 }; String birth() { @@ -482,38 +483,52 @@ public: return(""); } - // Invoke the entry unit through one of its directive handlers - // (render/cli/websocket/serve_http, selected by kind). handler_present, when - // provided, reports whether the unit actually exports that handler — the - // caller maps a missing render to an empty body (native parity) and a - // missing cli/serve handler to a 404. - String invoke_entry(const String& entry_source_path, int32_t kind, bool* handler_present = 0) + // Invoke the entry unit through a named handler ("render"/"cli"/"websocket"/ + // "serve_http:named"). The handler is just an export name — same resolve + + // ONCE + dispatch path as a component. handler_present, when provided, + // reports whether the unit exports it (caller maps a missing render to an + // empty body, a missing cli/serve handler to a 404). + String invoke_entry(const String& entry_source_path, const String& handler, bool* handler_present = 0) { entry_dir = dir_of(entry_source_path); size_t unit_index = 0; String error = load_unit(entry_source_path, unit_index); if(error != "") return(error); - String symbol = kind == RESOLVE_CLI ? "__uce_cli" : "__uce_render"; - bool present = (bool)unit_func(unit_index, symbol); + bool present = (bool)unit_func(unit_index, handler_export_symbol(handler)); if(handler_present) *handler_present = present; if(!present) return(""); - // Invoke through the core so the entry gets the same ONCE() + dispatch - // path as components (unit is pre-loaded above; resolution is cached). + // Invoke through the core: it resolves the same handler and runs ONCE + + // dispatch (unit is pre-loaded above; resolution is cached). The core + // entry takes two guest buffers — the unit path and the handler name. auto entry = core_func("uce_wasm_invoke_entry"); if(!entry) return("core does not export uce_wasm_invoke_entry"); - int32_t guest_ptr = 0; - error = call_core("uce_alloc", { (int32_t)entry_source_path.size() }, &guest_ptr); - if(error != "" || guest_ptr == 0) + int32_t path_ptr = 0, handler_ptr = 0; + error = call_core("uce_alloc", { (int32_t)entry_source_path.size() }, &path_ptr); + if(error != "" || path_ptr == 0) return(error == "" ? String("guest uce_alloc failed for entry path") : error); - error = guest_write((u32)guest_ptr, entry_source_path); + error = guest_write((u32)path_ptr, entry_source_path); + if(error == "") + { + error = call_core("uce_alloc", { (int32_t)handler.size() }, &handler_ptr); + if(error == "" && handler_ptr == 0) + error = "guest uce_alloc failed for handler"; + } + if(error == "") + error = guest_write((u32)handler_ptr, handler); if(error != "") + { + if(path_ptr) call_core("uce_free", { path_ptr }, 0); + if(handler_ptr) call_core("uce_free", { handler_ptr }, 0); return(error); - auto result = entry->call(ctx(), { wasmtime::Val(guest_ptr), wasmtime::Val((int32_t)entry_source_path.size()), wasmtime::Val(kind) }); - call_core("uce_free", { guest_ptr }, 0); + } + auto result = entry->call(ctx(), { wasmtime::Val(path_ptr), wasmtime::Val((int32_t)entry_source_path.size()), + wasmtime::Val(handler_ptr), wasmtime::Val((int32_t)handler.size()) }); + call_core("uce_free", { path_ptr }, 0); + call_core("uce_free", { handler_ptr }, 0); if(!result) return(trap_text(result.err())); return(""); @@ -521,7 +536,7 @@ public: String render_entry(const String& entry_source_path) { - return(invoke_entry(entry_source_path, RESOLVE_RENDER)); + return(invoke_entry(entry_source_path, "render")); } String collect(WasmResponse& response) @@ -891,6 +906,12 @@ private: return(stat(path.c_str(), &st) == 0 && S_ISREG(st.st_mode)); } + static bool dir_exists_host(const String& path) + { + struct stat st; + return(stat(path.c_str(), &st) == 0 && S_ISDIR(st.st_mode)); + } + String resolve_source_path(const String& file_name, const String& current_unit) { std::vector bases; @@ -926,7 +947,7 @@ private: // guest file access policy: only inside the site tree, resolved against // the entry unit's directory first (the native cwd convention), then the // site root; containment checked on the canonicalized path - String resolve_guest_file(const String& raw, const String& current_unit = "") + String resolve_guest_file(const String& raw, const String& current_unit = "", bool allow_dir = false) { if(raw == "" || raw.find('\0') != String::npos) return(""); @@ -970,7 +991,7 @@ private: } if(!allowed) continue; - if(file_exists_host(path)) + if(file_exists_host(path) || (allow_dir && dir_exists_host(path))) return(path); } return(""); @@ -1022,8 +1043,20 @@ private: return(result); } - // hostcall body: uce_host_component_resolve(target, kind, current) → slot - int32_t component_resolve(const String& target, int32_t kind, const String& current_unit, String& resolved_out) + // __uce_[_] for a handler spec like "component:CARD" / "render". + static String handler_export_symbol(const String& handler) + { + auto colon = handler.find(":"); + String symbol = "__uce_" + (colon == String::npos ? handler : handler.substr(0, colon)); + if(colon != String::npos) + symbol += "_" + sanitize_symbol_suffix(handler.substr(colon + 1)); + return(symbol); + } + + // hostcall body: uce_host_component_resolve(unit, handler, current) → slot. + // `handler` names the export ("render", "component:CARD", "cli", + // "serve_http:named", "once") or is "exists" (probe only, loads nothing). + int32_t component_resolve(const String& target, const String& handler, const String& current_unit, String& resolved_out) { auto probe_start = std::chrono::steady_clock::now(); auto record_probe = [&]() { @@ -1032,13 +1065,6 @@ private: std::chrono::steady_clock::now() - probe_start).count(); }; String file_name = target; - String render_name; - auto split = target.find(":"); - if(split != String::npos) - { - render_name = target.substr(split + 1); - file_name = target.substr(0, split); - } if(file_name == "" && current_unit != "") file_name = current_unit; if(file_name == "") @@ -1054,7 +1080,7 @@ private: return(0); } resolved_out = resolved; - if(kind == RESOLVE_EXISTS) + if(handler == "exists") { record_probe(); return(1); @@ -1068,12 +1094,7 @@ private: record_probe(); return(0); } - String symbol = kind == RESOLVE_RENDER ? "__uce_render" - : kind == RESOLVE_ONCE ? "__uce_once" - : kind == RESOLVE_CLI ? "__uce_cli" - : "__uce_component"; - if(render_name != "") - symbol += "_" + sanitize_symbol_suffix(render_name); + String symbol = handler_export_symbol(handler); String slot_key = resolved + ":" + symbol; auto cached = handler_slots.find(slot_key); if(cached != handler_slots.end()) @@ -1081,17 +1102,17 @@ private: record_probe(); return((int32_t)cached->second); } - auto handler = unit_func(unit_index, symbol); - if(!handler) + auto handler_fn = unit_func(unit_index, symbol); + if(!handler_fn) { // ONCE is optional per unit; a missing __uce_once is not an error - if(kind != RESOLVE_ONCE) + if(handler != "once") fprintf(stderr, "[wasm] %s does not export %s\n", resolved.c_str(), symbol.c_str()); record_probe(); return(0); } u32 slot = 0; - error = place_funcref(*handler, slot); + error = place_funcref(*handler_fn, slot); if(error != "") { fprintf(stderr, "[wasm] %s\n", error.c_str()); @@ -1184,6 +1205,31 @@ private: results[0] = Val(resolved != "" ? (int32_t)1 : (int32_t)0); return(std::monostate()); })); + if(mod == "env" && name == "uce_host_file_mkdir") + return(add([self](Caller, Span args, Span results) -> Result { + String path, current; + self->hostcall_read(args[0].i32(), args[1].i32(), path); + self->hostcall_read(args[2].i32(), args[3].i32(), current); + String resolved = self->resolve_guest_write(path, current); + int ok = 0; + if(resolved != "") + ok = (::mkdir(resolved.c_str(), 0777) == 0 || errno == EEXIST) ? 1 : 0; + results[0] = Val((int32_t)ok); + return(std::monostate()); + })); + if(mod == "env" && name == "uce_host_file_mtime") + return(add([self](Caller, Span args, Span results) -> Result { + String path, current; + self->hostcall_read(args[0].i32(), args[1].i32(), path); + self->hostcall_read(args[2].i32(), args[3].i32(), current); + String resolved = self->resolve_guest_file(path, current); + int64_t mtime = 0; + struct stat st; + if(resolved != "" && stat(resolved.c_str(), &st) == 0) + mtime = (int64_t)st.st_mtime; + results[0] = Val((int64_t)mtime); + return(std::monostate()); + })); if(mod == "env" && name == "uce_host_file_read") return(add([self](Caller, Span args, Span results) -> Result { String path, current; @@ -1204,6 +1250,38 @@ private: results[0] = Val((int32_t)bytes.size()); return(std::monostate()); })); + if(mod == "env" && name == "uce_host_file_list") + return(add([self](Caller, Span args, Span results) -> Result { + String path, current; + self->hostcall_read(args[0].i32(), args[1].i32(), path); + self->hostcall_read(args[2].i32(), args[3].i32(), current); + String resolved = self->resolve_guest_file(path, current, true /*allow_dir*/); + String listing; + if(resolved != "") + { + std::vector names; + if(DIR* d = opendir(resolved.c_str())) + { + while(struct dirent* e = readdir(d)) + { + String n = e->d_name; + if(n != "." && n != "..") + names.push_back(n); + } + closedir(d); + } + // match the native ls -1 convention: bare names, sorted + std::sort(names.begin(), names.end()); + listing = join(names, "\n"); + } + u32 cap = (u32)args[5].i32(); + int32_t buf = args[4].i32(); + // length-query convention: no copy unless the buffer fits + if(buf != 0 && cap >= listing.size()) + self->hostcall_write(buf, listing); + results[0] = Val((int32_t)listing.size()); + return(std::monostate()); + })); if(mod == "env" && name == "uce_host_file_write") return(add([self](Caller, Span args, Span results) -> Result { String path, current, content; @@ -1714,17 +1792,18 @@ private: })); if(mod == "env" && name == "uce_host_component_resolve") return(add([self](Caller, Span args, Span results) -> Result { - String target, current, resolved; + String target, handler, current, resolved; self->hostcall_read(args[0].i32(), args[1].i32(), target); - self->hostcall_read(args[3].i32(), args[4].i32(), current); - int32_t slot = self->component_resolve(target, args[2].i32(), current, resolved); - u32 cap = (u32)args[6].i32(); + self->hostcall_read(args[2].i32(), args[3].i32(), handler); + self->hostcall_read(args[4].i32(), args[5].i32(), current); + int32_t slot = self->component_resolve(target, handler, current, resolved); + u32 cap = (u32)args[7].i32(); if(cap > 0) { if(resolved.size() >= cap) resolved = resolved.substr(0, cap - 1); resolved.push_back('\0'); - self->hostcall_write(args[5].i32(), resolved); + self->hostcall_write(args[6].i32(), resolved); } results[0] = Val(slot); return(std::monostate()); @@ -1752,7 +1831,7 @@ private: // ---- public entry: one request through one workspace ----------------------- inline WasmResponse wasm_worker_serve(WasmWorker& worker, const DValue& context_tree, const String& entry_source_path, - int32_t kind = WasmWorkspace::RESOLVE_RENDER) + const String& handler = "render") { WasmResponse response; WasmWorkspace workspace(worker); @@ -1763,7 +1842,7 @@ inline WasmResponse wasm_worker_serve(WasmWorker& worker, const DValue& context_ if(error == "") error = workspace.apply_context(context_tree); if(error == "") - error = workspace.invoke_entry(entry_source_path, kind, &response.handler_present); + error = workspace.invoke_entry(entry_source_path, handler, &response.handler_present); if(error == "") error = workspace.collect(response); response.workspace_birth_us = workspace.workspace_birth_us;