From 72a37b178c482329964f0bd3f44e9bdb5cf8955f Mon Sep 17 00:00:00 2001 From: Steve Biedermann Date: Tue, 3 Nov 2026 15:36:21 +0100 Subject: [PATCH] feat!: use runa-debug WebSockets with live image notifications Replace plain TCP JSON-lines connections with bounded loopback WebSockets using the required runa-debug subprotocol. Keep single-threaded image dispatch and stdio framing unchanged. Broadcast coalesced, non-destructive image_changed invalidations after successful mutations. Update the REPL client and automatically refresh VS Code actor trees and open inspectors, preserving request correlation and connection-session isolation. Validate negotiation, message limits and fragmentation, multi-client notifications, real REPL interoperability, and automatic mailbox/lifecycle/failure inspection in a VS Code extension host. BREAKING CHANGE: the network endpoint now accepts only runa-debug WebSockets at the root path, not plain TCP JSON Lines. Rust network integrations must use WebSocketClient and serve_websocket. Request/response DTOs remain protocol v3; stdio is unchanged. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- AGENTS.md | 6 +- CHANGELOG.md | 7 + Cargo.lock | 293 ++++++++++++++++++++++- README.md | 28 ++- crates/runa-client/Cargo.toml | 1 + crates/runa-client/src/lib.rs | 250 ++++++++++++++++--- crates/runa-protocol/src/lib.rs | 15 ++ crates/runa-server/Cargo.toml | 3 + crates/runa-server/src/lib.rs | 224 ++--------------- crates/runa-server/src/main.rs | 4 +- crates/runa-server/src/network.rs | 175 ++++++++++++++ crates/runa-server/tests/local_client.rs | 235 ++++++++++++++++-- crates/runa/Cargo.toml | 1 + crates/runa/src/main.rs | 50 +++- docs/architecture.md | 32 +-- docs/protocol.md | 85 +++++-- docs/runtime.md | 14 +- mise.toml | 2 +- vscode/README.md | 23 +- vscode/integration/host.ts | 34 ++- vscode/integration/repl.test.ts | 73 ++++++ vscode/integration/server.test.ts | 116 +++++---- vscode/integration/server.ts | 43 ++-- vscode/integration/websocket.test.ts | 60 +++++ vscode/package-lock.json | 35 ++- vscode/package.json | 6 +- vscode/src/client.ts | 134 ++++++----- vscode/src/extension.ts | 33 ++- vscode/src/protocol.ts | 36 ++- vscode/src/refresh.ts | 40 ++++ vscode/test/client.test.ts | 108 +++++++-- vscode/test/fixtures.ts | 32 ++- vscode/test/protocol.test.ts | 36 +++ vscode/test/refresh.test.ts | 53 ++++ 34 files changed, 1761 insertions(+), 526 deletions(-) create mode 100644 crates/runa-server/src/network.rs create mode 100644 vscode/integration/repl.test.ts create mode 100644 vscode/integration/websocket.test.ts create mode 100644 vscode/src/refresh.ts create mode 100644 vscode/test/refresh.test.ts diff --git a/AGENTS.md b/AGENTS.md index 74eae8a..e889ebd 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -68,8 +68,8 @@ proposed capabilities as plans, not existing behavior. - `crates/runa-runtime/`: single-worker scheduling and in-memory transitions. - `crates/runa-image-api/`: trusted in-process services for operating on an image. - `crates/runa-protocol/`: versioned transport-neutral request, response, and event DTOs. -- `crates/runa-server/`: protocol dispatch and local JSON-lines stdio transport. -- `crates/runa-client/`: local child-process client for the JSON-lines protocol. +- `crates/runa-server/`: protocol dispatch, JSON-lines stdio, and loopback WebSockets. +- `crates/runa-client/`: stdio child-process and runa-debug WebSocket clients. - `runa-std/actors/`: default Rune actor scripts loaded by the local CLI. - `docs/architecture.md`: architecture, roadmap, risks, and acceptance scenario. - `docs/decisions/`: architectural decisions. @@ -103,7 +103,7 @@ are required; install the locked extension dependencies with `npm --prefix vscode ci`. The workspace tests cover execution probes, actor runtime invariants, and the -interactive CLI. Extension tests include a real Rust-server TCP round trip. +interactive CLI. Extension tests include a real Rust-server WebSocket round trip. `mise run start` opens the local CLI. The separate `mise run test-extension-host` task verifies inspector commands in VS Code and requires a display (or `xvfb-run -a mise run test-extension-host` on headless Linux). diff --git a/CHANGELOG.md b/CHANGELOG.md index 2d59f34..79aef6c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,10 @@ All notable changes to this project will be documented in this file. ### Added +- Non-destructive `image_changed` server notifications with coalesced + per-connection revisions and automatic VS Code tree/inspector refresh. +- Tests for WebSocket subprotocol negotiation, invalid messages, multi-client + notifications, and automatic mailbox/lifecycle/failure inspector updates. - Initial TypeScript VS Code extension with explicit localhost connection management, an actor explorer, and read-only state/mailbox/execution snapshots. - Bounded protocol-v3 extension client with lossless 64-bit JSON integers, @@ -86,6 +90,9 @@ All notable changes to this project will be documented in this file. ### Changed +- Replaced the plain TCP JSON-lines transport with loopback-only WebSockets + using the required `runa-debug` subprotocol in the server, REPL, and VS Code. + Stdio framing and version 3 request/response shapes are unchanged. - Renamed the local REPL mise task from `run` to `start`. - Protocol v3 and Rune's two-argument `handle(state, message)` replace the temporary integer-only state and kind/value/target_ref message envelope. diff --git a/Cargo.lock b/Cargo.lock index d09080f..068535c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -57,9 +57,24 @@ dependencies = [ "cc", "cfg-if", "constant_time_eq", - "cpufeatures", + "cpufeatures 0.3.1", ] +[[package]] +name = "block-buffer" +version = "0.10.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3078c7629b62d3f0439517fa394996acacc5cbc91c5a20d8c658e77abd503a71" +dependencies = [ + "generic-array", +] + +[[package]] +name = "bytes" +version = "1.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04" + [[package]] name = "cc" version = "1.10.0" @@ -107,6 +122,15 @@ version = "0.4.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3d52eff69cd5e647efe296129160853a42795992097e8af39800e1060caeea9b" +[[package]] +name = "cpufeatures" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "59ed5838eebb26a2bb2e58f6d5b5316989ae9d08bab10e0e6d103e656d1b0280" +dependencies = [ + "libc", +] + [[package]] name = "cpufeatures" version = "0.3.1" @@ -122,6 +146,32 @@ version = "1.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "790eea4361631c5e7d22598ecd5723ff611904e3344ce8720784c93e3d83d40b" +[[package]] +name = "crypto-common" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "78c8292055d1c1df0cce5d180393dc8cce0abec0a7102adb6c7b1eef6016d60a" +dependencies = [ + "generic-array", + "typenum", +] + +[[package]] +name = "data-encoding" +version = "2.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4583a4551df46e2792f82ceeac45e850d2e2d5debba0b91f102385cda5b11f06" + +[[package]] +name = "digest" +version = "0.10.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292" +dependencies = [ + "block-buffer", + "crypto-common", +] + [[package]] name = "endian-type" version = "0.1.2" @@ -167,6 +217,23 @@ version = "0.3.34" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "92d699e522242e69e3003b94ecc1f960f3a5e015aa7c5d7486e65ad01dd94f5e" +[[package]] +name = "futures-macro" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9fb9654ba8355388abeb8dcb4fc62f511300867002afc858860463bdd9fe0c44" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.9", +] + +[[package]] +name = "futures-sink" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1944426bf7d03f1d14f708785e4b33efd750b36d48a157b836b3efc15ede8e1d" + [[package]] name = "futures-task" version = "0.3.34" @@ -180,6 +247,8 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0d50a92467f8ba5dd6e3ee5d4bd04d73ab2e4e1c44474a0674821dfce14b79bc" dependencies = [ "futures-core", + "futures-macro", + "futures-sink", "futures-task", "pin-project-lite", "slab", @@ -200,6 +269,16 @@ dependencies = [ "windows-result", ] +[[package]] +name = "generic-array" +version = "0.14.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "85649ca51fd72272d7821adaf274ad91c288277713d9c18820d8499a7ff69e9a" +dependencies = [ + "typenum", + "version_check", +] + [[package]] name = "getrandom" version = "0.2.17" @@ -211,6 +290,18 @@ dependencies = [ "wasi", ] +[[package]] +name = "getrandom" +version = "0.3.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "899def5c37c4fd7b2664648c28120ecec138e4d395b459e5ca34f9cce2dd77fd" +dependencies = [ + "cfg-if", + "libc", + "r-efi", + "wasip2", +] + [[package]] name = "home" version = "0.5.12" @@ -220,6 +311,22 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "http" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "918d3568bebf352712bc2ef3d46a8bcf1a75b373be6539de198e9105cbbf9ce0" +dependencies = [ + "bytes", + "itoa", +] + +[[package]] +name = "httparse" +version = "1.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6dbf3de79e51f3d586ab4cb9d5c3e2c14aa28ed23d180cf89b4df0454a69cc87" + [[package]] name = "itoa" version = "1.0.18" @@ -278,6 +385,17 @@ version = "2.8.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98" +[[package]] +name = "mio" +version = "1.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1788edb87fdc09c7e26304471e2f5be8cdefb1b6930d6e3985fc02ff53bf86ee" +dependencies = [ + "libc", + "wasi", + "windows-sys 0.61.2", +] + [[package]] name = "musli" version = "0.0.124" @@ -316,7 +434,7 @@ version = "0.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3ffa00dec017b5b1a8b7cf5e2c008bfda1aa7e0697ac1508b491fdf2622fb4d8" dependencies = [ - "rand", + "rand 0.8.8", ] [[package]] @@ -490,6 +608,12 @@ dependencies = [ "proc-macro2", ] +[[package]] +name = "r-efi" +version = "5.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "69cdb34c158ceb288df11e18b4bd39de994f6657d83847bdffdbd7f346754b0f" + [[package]] name = "radix_trie" version = "0.2.1" @@ -507,8 +631,18 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e058c7de0b26af77780c769414d6257830bb240f3c38477dbc2c16e5f54d6d4c" dependencies = [ "libc", - "rand_chacha", - "rand_core", + "rand_chacha 0.3.1", + "rand_core 0.6.4", +] + +[[package]] +name = "rand" +version = "0.9.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b9ef1d0d795eb7d84685bca4f72f3649f064e6641543d3a8c415898726a57b41" +dependencies = [ + "rand_chacha 0.9.0", + "rand_core 0.9.5", ] [[package]] @@ -518,7 +652,17 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e6c10a63a0fa32252be49d21e7709d4d4baf8d231c2dbce1eaa8141b9b127d88" dependencies = [ "ppv-lite86", - "rand_core", + "rand_core 0.6.4", +] + +[[package]] +name = "rand_chacha" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d3022b5f1df60f26e1ffddd6c66e8aa15de382ae63b3a0c1bfc0e4d3e3f325cb" +dependencies = [ + "ppv-lite86", + "rand_core 0.9.5", ] [[package]] @@ -527,7 +671,16 @@ version = "0.6.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ec0be4795e2f6a28069bec0b5ff3e2ac9bafc99e6a9a7dc3547996c5c816922c" dependencies = [ - "getrandom", + "getrandom 0.2.17", +] + +[[package]] +name = "rand_core" +version = "0.9.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "76afc826de14238e6e8c374ddcc1fa19e374fd8dd986b0d2af0d02377261d83c" +dependencies = [ + "getrandom 0.3.4", ] [[package]] @@ -557,6 +710,7 @@ dependencies = [ "runa-server", "rustyline", "serde_json", + "tungstenite", ] [[package]] @@ -572,6 +726,7 @@ version = "0.0.0" dependencies = [ "runa-protocol", "serde_json", + "tungstenite", ] [[package]] @@ -634,12 +789,15 @@ dependencies = [ name = "runa-server" version = "0.0.0" dependencies = [ + "futures-util", "runa-actor", "runa-client", "runa-code", "runa-image-api", "runa-protocol", "serde_json", + "tokio", + "tokio-tungstenite", ] [[package]] @@ -831,6 +989,17 @@ dependencies = [ "zmij", ] +[[package]] +name = "sha1" +version = "0.10.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a978451301f4db1d02937a4ab3ccce137717b81826e79b7d49ffe3244a13c3b8" +dependencies = [ + "cfg-if", + "cpufeatures 0.2.17", + "digest", +] + [[package]] name = "sharded-slab" version = "0.1.7" @@ -864,6 +1033,16 @@ version = "1.16.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d0d5d4b1981b8bbbb7aed204cae5726d683d5d6ac7e74ec54122ea28f315e7c8" +[[package]] +name = "socket2" +version = "0.6.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c3d1e2c7f27f8d4cb10542a02c49005dbd6e93095799d6f3be745fae9f8fedd4" +dependencies = [ + "libc", + "windows-sys 0.61.2", +] + [[package]] name = "syn" version = "2.0.119" @@ -901,6 +1080,26 @@ dependencies = [ "winapi-util", ] +[[package]] +name = "thiserror" +version = "2.0.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09e52cb86a36cede5cb101bf8908837b3e4c6e5e59fe7fd85c23fb56200d189e" +dependencies = [ + "thiserror-impl", +] + +[[package]] +name = "thiserror-impl" +version = "2.0.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fe5197923287db20a58125f0bc85c062f7f2c892de97b18c356f9efb14b28524" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.9", +] + [[package]] name = "thread_local" version = "1.1.10" @@ -910,6 +1109,44 @@ dependencies = [ "cfg-if", ] +[[package]] +name = "tokio" +version = "1.54.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b52fd3255258d26fac7264ab5a76b53c13367136dc011d3d6ad4f8ed070396a3" +dependencies = [ + "bytes", + "libc", + "mio", + "pin-project-lite", + "socket2", + "tokio-macros", + "windows-sys 0.61.2", +] + +[[package]] +name = "tokio-macros" +version = "2.7.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "78773a2a397f451582ce068015985c33193cf6dea8b74d2a639fe457b2f07b0e" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.9", +] + +[[package]] +name = "tokio-tungstenite" +version = "0.28.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d25a406cddcc431a75d3d9afc6a7c0f7428d4891dd973e4d54c56b46127bf857" +dependencies = [ + "futures-util", + "log", + "tokio", + "tungstenite", +] + [[package]] name = "tracing" version = "0.1.44" @@ -959,12 +1196,35 @@ dependencies = [ "tracing-log", ] +[[package]] +name = "tungstenite" +version = "0.28.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8628dcc84e5a09eb3d8423d6cb682965dea9133204e8fb3efee74c2a0c259442" +dependencies = [ + "bytes", + "data-encoding", + "http", + "httparse", + "log", + "rand 0.9.5", + "sha1", + "thiserror", + "utf-8", +] + [[package]] name = "twox-hash" version = "2.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "86a801b3cea342a06d468c8710662aa29e5e05e4f5c0d62f00bbb7f2ad7941c2" +[[package]] +name = "typenum" +version = "1.20.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6f5e870be6c3b371b77fe0ee0bafb859fa4964b4404c27de1d380043c4dda20" + [[package]] name = "unicode-ident" version = "1.0.27" @@ -989,6 +1249,12 @@ version = "0.2.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ca718471876992059f251e33aafc334346ada08042b7661dd1fe313a9bced557" +[[package]] +name = "utf-8" +version = "0.7.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09cc8ee72d2a9becf2f2febe0205bbed8fc6615b7cb429ad062dc7b7ddd036a9" + [[package]] name = "utf8parse" version = "0.2.2" @@ -1013,6 +1279,15 @@ version = "0.11.1+wasi-snapshot-preview1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b" +[[package]] +name = "wasip2" +version = "1.0.4+wasi-0.2.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b67efb37e106e55ce722a510d6b5f9c17f083e5fc79afc2badeb12cc313d9487" +dependencies = [ + "wit-bindgen", +] + [[package]] name = "winapi-util" version = "0.1.11" @@ -1119,6 +1394,12 @@ version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec" +[[package]] +name = "wit-bindgen" +version = "0.57.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1ebf944e87a7c253233ad6766e082e3cd714b5d03812acc24c318f549614536e" + [[package]] name = "zerocopy" version = "0.8.63" diff --git a/README.md b/README.md index f8cfc12..3e9f268 100644 --- a/README.md +++ b/README.md @@ -22,9 +22,10 @@ registry, and the CLI exposes publishing, history, rollback, safe-boundary installation, pause/resume, and cancellation through a trusted in-process image API. A versioned transport-neutral protocol contract is now defined, but no authentication is implemented. The dispatcher is available through local -JSON-lines stdio and an unauthenticated TCP listener restricted to localhost; -both use a bounded request size. The interactive CLI connects to the localhost -server over TCP, so multiple clients operate on the same server-owned image. +JSON-lines stdio and an unauthenticated WebSocket listener restricted to localhost; +both use a bounded request size. The interactive CLI and VS Code connect with +the `runa-debug` WebSocket subprotocol, so clients share one server-owned image. +Server change notifications automatically refresh VS Code's actor tree and inspectors. The prototype uses worker-local invocation stacks and cooperative instruction slices, and actor snapshots report basic active-execution progress. Native @@ -70,22 +71,22 @@ For a local child-process protocol endpoint, run `mise run serve` and exchange one JSON request per line on stdin; responses are written to stdout. Request lines are limited to 1 MiB. -For a TCP endpoint, run `mise run serve-network`. It listens on -`127.0.0.1:3737` and accepts the same JSON-lines protocol, with up to 16 +For a WebSocket endpoint, run `mise run serve-network`. It listens on +`ws://127.0.0.1:3737/` and requires the `runa-debug` subprotocol, with up to 16 simultaneous clients sharing one live image. The listener rejects non-loopback addresses and has no authentication; any local process that can connect has full access to the development API. Use it only on a trusted local machine. The `runa-client` crate can launch the stdio process or connect over loopback -TCP and exchange typed protocol requests with request/response correlation. -TCP connection establishment is limited to 5 seconds and reads/writes to 30 +WebSockets and exchange typed protocol requests with request/response correlation. +Network connection establishment is limited to 5 seconds and response waits/writes to 30 seconds. Transport failures disconnect the client and are never automatically retried: an operation may have committed even if its response was lost. The VS Code extension connects over the network to this same live image; actors remain in the image regardless of how clients connect. Remote network access requires authentication and authorization, which are not implemented. -The TCP server publishes the bundled `counter` behavior once at startup; +The WebSocket server publishes the bundled `counter` behavior once at startup; connecting a client does not publish or reset behavior revisions: ```text @@ -132,11 +133,12 @@ open this repository and launch **Run Runa Extension** from Run and Debug (F5). Start `mise run serve-network`, then use **Runa: Connect to Local Image** in the Extension Development Host. The **Runa Actors** view in Explorer lists actors created through the REPL or another client. Click an actor to inspect -it, and use **Runa: Refresh Actors** to fetch current snapshots. +it. Server notifications automatically refresh the tree and open snapshots; +**Runa: Refresh Actors** is also available on demand. The extension does not start, tick, or stop the image. Disconnecting leaves -actors running in the server. Editing, spawn/send controls, and event -subscriptions are not implemented yet. See the [extension guide](vscode/README.md) +actors running in the server. Editing, spawn/send controls, and detailed runtime +event subscriptions are not implemented yet. See the [extension guide](vscode/README.md) for connection restrictions, snapshot semantics, and tests. ## Development @@ -160,7 +162,7 @@ Tombi formats TOML files offline; rustfmt formats all workspace Rust code; Prettier formats the extension. Clippy checks all workspace targets with warnings treated as errors. Extension checks include strict TypeScript and Prettier. The test task runs the Rust workspace tests, extension unit tests, -and a TypeScript-to-Rust TCP interoperability test. +and a TypeScript-to-Rust WebSocket interoperability test. `mise run test-extension-host` additionally downloads VS Code 1.96.4 into an ignored test cache and tests the actual inspector commands in an isolated Extension Development Host. It requires a graphical display; on headless Linux, @@ -181,7 +183,7 @@ use `xvfb-run -a mise run test-extension-host`. │ ├── runa-image-api/ # Trusted in-process image operations │ ├── runa-protocol/ # Versioned transport-neutral wire DTOs │ ├── runa-server/ # Protocol dispatcher and local JSON-lines transport -│ └── runa-client/ # Stdio child-process and loopback TCP clients +│ └── runa-client/ # Stdio child-process and loopback WebSocket clients ├── docs/ │ ├── architecture.md │ ├── protocol.md diff --git a/crates/runa-client/Cargo.toml b/crates/runa-client/Cargo.toml index 3fb5d2d..f6a321c 100644 --- a/crates/runa-client/Cargo.toml +++ b/crates/runa-client/Cargo.toml @@ -6,3 +6,4 @@ edition = "2024" [dependencies] runa-protocol = { path = "../runa-protocol" } serde_json = "1" +tungstenite = "0.28" diff --git a/crates/runa-client/src/lib.rs b/crates/runa-client/src/lib.rs index ca99699..2b1ff68 100644 --- a/crates/runa-client/src/lib.rs +++ b/crates/runa-client/src/lib.rs @@ -1,13 +1,18 @@ -//! Stdio and loopback TCP clients for the versioned JSON-lines protocol. +//! Stdio and loopback WebSocket clients for the versioned protocol. use std::ffi::OsStr; use std::fmt; use std::io::{self, BufRead, BufReader, BufWriter, Write}; use std::net::{Shutdown, SocketAddr, TcpStream}; use std::process::{Child, ChildStdin, ChildStdout, Command, ExitStatus, Stdio}; -use std::time::Duration; +use std::time::{Duration, Instant}; -use runa_protocol::{Command as ProtocolCommand, PROTOCOL_VERSION, Request, Response}; +use runa_protocol::{ + Command as ProtocolCommand, ImageEvent, Notification, PROTOCOL_VERSION, Request, Response, + WEBSOCKET_SUBPROTOCOL, +}; +use tungstenite::client::IntoClientRequest; +use tungstenite::{Message, WebSocket, protocol::WebSocketConfig}; pub const MAX_MESSAGE_LINE_BYTES: usize = 1024 * 1024; @@ -22,12 +27,14 @@ pub struct ProtocolClient { next_request_id: u64, } -pub struct TcpClient { - client: ProtocolClient, BufWriter>, +pub struct WebSocketClient { + socket: WebSocket, + next_request_id: u64, + revision: u64, disconnected: bool, } -impl TcpClient { +impl WebSocketClient { /// Connect only to loopback, without authentication or automatic retries. pub fn connect(address: SocketAddr) -> io::Result { if !address.ip().is_loopback() { @@ -37,13 +44,43 @@ impl TcpClient { )); } let stream = TcpStream::connect_timeout(&address, Duration::from_secs(5))?; - stream.set_read_timeout(Some(Duration::from_secs(30)))?; - stream.set_write_timeout(Some(Duration::from_secs(30)))?; + stream.set_read_timeout(Some(Duration::from_secs(5)))?; + stream.set_write_timeout(Some(Duration::from_secs(5)))?; + let mut request = format!("ws://{address}/") + .into_client_request() + .map_err(io::Error::other)?; + request.headers_mut().insert( + "sec-websocket-protocol", + WEBSOCKET_SUBPROTOCOL.parse().map_err(io::Error::other)?, + ); + let config = WebSocketConfig::default() + .max_message_size(Some(MAX_MESSAGE_LINE_BYTES)) + .max_frame_size(Some(MAX_MESSAGE_LINE_BYTES)) + .max_write_buffer_size(2 * MAX_MESSAGE_LINE_BYTES); + let (socket, response) = + tungstenite::client::client_with_config(request, stream, Some(config)) + .map_err(io::Error::other)?; + if response + .headers() + .get("sec-websocket-protocol") + .and_then(|value| value.to_str().ok()) + != Some(WEBSOCKET_SUBPROTOCOL) + { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "server did not negotiate runa-debug", + )); + } + socket + .get_ref() + .set_read_timeout(Some(Duration::from_secs(30)))?; + socket + .get_ref() + .set_write_timeout(Some(Duration::from_secs(30)))?; Ok(Self { - client: ProtocolClient::new( - BufReader::new(stream.try_clone()?), - BufWriter::new(stream), - ), + socket, + next_request_id: 0, + revision: 0, disconnected: false, }) } @@ -52,19 +89,89 @@ impl TcpClient { if self.disconnected { return Err(io::Error::new( io::ErrorKind::NotConnected, - "TCP client is disconnected; reconnect explicitly", + "WebSocket client is disconnected; reconnect explicitly", ) .into()); } - let result = self.client.request(command); + let result = self.request_inner(command); if result.is_err() { self.disconnected = true; // A lost response may follow a committed operation. Never retry it // or reuse a stream whose request/response alignment is uncertain. - let _ = self.client.writer.get_ref().shutdown(Shutdown::Both); + let _ = self.socket.get_ref().shutdown(Shutdown::Both); } result } + + fn request_inner(&mut self, command: ProtocolCommand) -> Result { + self.next_request_id = self + .next_request_id + .checked_add(1) + .ok_or(ClientError::RequestIdsExhausted)?; + let request_id = format!("runa-client-{}", self.next_request_id); + let encoded = serde_json::to_string(&Request { + version: PROTOCOL_VERSION, + request_id: request_id.clone(), + command, + })?; + if encoded.len() > MAX_MESSAGE_LINE_BYTES { + return Err(ClientError::RequestTooLarge); + } + self.socket + .send(Message::Text(encoded.into())) + .map_err(io::Error::other)?; + let deadline = Instant::now() + Duration::from_secs(30); + loop { + let remaining = deadline + .checked_duration_since(Instant::now()) + .filter(|duration| !duration.is_zero()) + .ok_or_else(|| { + io::Error::new(io::ErrorKind::TimedOut, "WebSocket response timed out") + })?; + self.socket.get_ref().set_read_timeout(Some(remaining))?; + match self.socket.read().map_err(io::Error::other)? { + Message::Text(text) => { + let value: serde_json::Value = serde_json::from_str(&text)?; + if value.get("event").is_some() { + let event: Notification = serde_json::from_value(value)?; + if event.version != PROTOCOL_VERSION { + return Err(ClientError::UnsupportedVersion(event.version)); + } + let ImageEvent::ImageChanged { revision } = event.event; + if revision <= self.revision { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "non-increasing image notification revision", + ) + .into()); + } + self.revision = revision; + continue; + } + let response: Response = serde_json::from_value(value)?; + if response.version != PROTOCOL_VERSION { + return Err(ClientError::UnsupportedVersion(response.version)); + } + if response.request_id != request_id { + return Err(ClientError::CorrelationMismatch { + expected: request_id, + actual: response.request_id, + }); + } + return Ok(response); + } + Message::Ping(_) => self.socket.flush().map_err(io::Error::other)?, + Message::Pong(_) => {} + _ => { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "expected a WebSocket text protocol message", + ) + .into()); + } + } + } + } } #[derive(Debug)] @@ -267,30 +374,115 @@ mod tests { use runa_protocol::{ProtocolError, Reply}; use std::io::Cursor; + struct DebugProtocol; + impl tungstenite::handshake::server::Callback for DebugProtocol { + fn on_request( + self, + _: &tungstenite::handshake::server::Request, + mut response: tungstenite::handshake::server::Response, + ) -> Result< + tungstenite::handshake::server::Response, + tungstenite::handshake::server::ErrorResponse, + > { + response.headers_mut().insert( + "sec-websocket-protocol", + WEBSOCKET_SUBPROTOCOL.parse().unwrap(), + ); + Ok(response) + } + } + + fn websocket_peer( + handler: impl FnOnce(WebSocket, Request) + Send + 'static, + ) -> (SocketAddr, std::thread::JoinHandle<()>) { + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let address = listener.local_addr().unwrap(); + let peer = std::thread::spawn(move || { + let (stream, _) = listener.accept().unwrap(); + let mut socket = tungstenite::accept_hdr(stream, DebugProtocol).unwrap(); + let message = socket.read().unwrap(); + let request = serde_json::from_str(message.to_text().unwrap()).unwrap(); + handler(socket, request); + }); + (address, peer) + } + #[test] - fn tcp_client_rejects_non_loopback_addresses() { - let error = TcpClient::connect("192.0.2.1:3737".parse().unwrap()) + fn websocket_client_validates_notifications_without_losing_response_correlation() { + let (address, peer) = websocket_peer(|mut socket, request| { + socket + .send(Message::Text( + serde_json::to_string(&Notification { + version: PROTOCOL_VERSION, + event: ImageEvent::ImageChanged { revision: 42 }, + }) + .unwrap() + .into(), + )) + .unwrap(); + socket + .send(Message::Text( + serde_json::to_string(&Response { + version: PROTOCOL_VERSION, + request_id: request.request_id, + result: Ok(Reply::Actors(Vec::new())), + }) + .unwrap() + .into(), + )) + .unwrap(); + }); + let mut client = WebSocketClient::connect(address).unwrap(); + assert_eq!( + client.request(ProtocolCommand::ListActors).unwrap().result, + Ok(Reply::Actors(Vec::new())) + ); + assert_eq!(client.revision, 42); + peer.join().unwrap(); + } + + #[test] + fn websocket_client_rejects_bad_versions_correlations_and_notifications() { + for message in [ + serde_json::json!({"version": 2, "request_id": "runa-client-1", + "result": {"Ok": {"kind": "actors", "data": []}}}), + serde_json::json!({"version": 3, "request_id": "wrong", + "result": {"Ok": {"kind": "actors", "data": []}}}), + serde_json::json!({"version": 3, "event": {"event": "image_changed", "data": {"revision": 0}}}), + serde_json::json!({"version": 3, "event": {"event": "unknown"}}), + ] { + let (address, peer) = websocket_peer(move |mut socket, _| { + socket + .send(Message::Text(message.to_string().into())) + .unwrap(); + }); + let mut client = WebSocketClient::connect(address).unwrap(); + assert!(client.request(ProtocolCommand::ListActors).is_err()); + assert!(matches!(client.request(ProtocolCommand::ListActors), + Err(ClientError::Io(error)) if error.kind() == io::ErrorKind::NotConnected)); + peer.join().unwrap(); + } + } + + #[test] + fn websocket_client_rejects_non_loopback_addresses() { + let error = WebSocketClient::connect("192.0.2.1:3737".parse().unwrap()) .err() .expect("non-loopback connection must be rejected"); assert_eq!(error.kind(), io::ErrorKind::InvalidInput); } #[test] - fn tcp_client_reports_disconnect_and_does_not_retry() { - let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); - let address = listener.local_addr().unwrap(); - let peer = std::thread::spawn(move || { - let (stream, _) = listener.accept().unwrap(); - let mut reader = BufReader::new(stream); - let mut line = String::new(); - reader.read_line(&mut line).unwrap(); - let request: Request = serde_json::from_str(&line).unwrap(); + fn websocket_client_reports_disconnect_and_does_not_retry() { + let (address, peer) = websocket_peer(|_socket, request| { assert_eq!(request.command, ProtocolCommand::ListActors); // Close after receiving the request, without sending a response. }); - let mut client = TcpClient::connect(address).unwrap(); - assert!(matches!(client.request(ProtocolCommand::ListActors), - Err(ClientError::Io(error)) if error.kind() == io::ErrorKind::UnexpectedEof)); + let mut client = WebSocketClient::connect(address).unwrap(); + assert!(matches!( + client.request(ProtocolCommand::ListActors), + Err(ClientError::Io(_)) + )); assert!(matches!(client.request(ProtocolCommand::ListActors), Err(ClientError::Io(error)) if error.kind() == io::ErrorKind::NotConnected)); peer.join().unwrap(); diff --git a/crates/runa-protocol/src/lib.rs b/crates/runa-protocol/src/lib.rs index 08f858c..4756979 100644 --- a/crates/runa-protocol/src/lib.rs +++ b/crates/runa-protocol/src/lib.rs @@ -4,6 +4,21 @@ pub use runa_core::Value; use serde::{Deserialize, Serialize}; pub const PROTOCOL_VERSION: u16 = 3; +pub const WEBSOCKET_SUBPROTOCOL: &str = "runa-debug"; + +/// Coalescible invalidation, not a drain of the image's runtime event queue. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct Notification { + pub version: u16, + pub event: ImageEvent, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(tag = "event", content = "data", rename_all = "snake_case")] +pub enum ImageEvent { + ImageChanged { revision: u64 }, +} #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[serde(deny_unknown_fields)] diff --git a/crates/runa-server/Cargo.toml b/crates/runa-server/Cargo.toml index 1266400..4ac591a 100644 --- a/crates/runa-server/Cargo.toml +++ b/crates/runa-server/Cargo.toml @@ -4,11 +4,14 @@ version = "0.0.0" edition = "2024" [dependencies] +futures-util = { version = "0.3", features = ["sink"] } runa-actor = { path = "../runa-actor" } runa-code = { path = "../runa-code" } runa-image-api = { path = "../runa-image-api" } runa-protocol = { path = "../runa-protocol" } serde_json = "1" +tokio = { version = "1", features = ["macros", "net", "rt", "sync", "time"] } +tokio-tungstenite = "0.28" [dev-dependencies] runa-client = { path = "../runa-client" } diff --git a/crates/runa-server/src/lib.rs b/crates/runa-server/src/lib.rs index fdd709b..46934c3 100644 --- a/crates/runa-server/src/lib.rs +++ b/crates/runa-server/src/lib.rs @@ -1,10 +1,11 @@ //! Transport-neutral dispatch from protocol requests to image API operations. -use std::io::{self, BufRead, Read, Write}; -use std::net::{SocketAddr, TcpListener, TcpStream}; +use std::io::{self, BufRead, Write}; +use std::net::{SocketAddr, TcpListener}; use std::num::NonZeroUsize; -use std::thread; -use std::time::Duration; + +mod network; +pub use network::serve_websocket; use runa_actor::{ActorRef, Event, Status}; use runa_image_api::ImageApi; @@ -14,7 +15,6 @@ use runa_protocol::{ }; pub const MAX_REQUEST_LINE_BYTES: usize = 1024 * 1024; -const MAX_NETWORK_CLIENTS: usize = 16; pub struct Server { image: ImageApi, @@ -44,171 +44,6 @@ pub fn bind_loopback_listener(address: SocketAddr) -> io::Result { TcpListener::bind(address) } -/// Serve JSON-lines requests from loopback clients using one shared image. -/// -/// The nonblocking event loop keeps the non-`Send` Rune runtime on one thread. -/// At most `MAX_NETWORK_CLIENTS` are active. -pub fn serve_tcp(listener: TcpListener, mut server: Server) -> io::Result<()> { - let address = listener.local_addr()?; - if !address.ip().is_loopback() { - return Err(io::Error::new( - io::ErrorKind::InvalidInput, - "network listener must be bound to a loopback address", - )); - } - listener.set_nonblocking(true)?; - eprintln!("runa-server: listening on {address}"); - let mut clients = Vec::new(); - loop { - loop { - match listener.accept() { - Ok((stream, peer_address)) => { - if clients.len() >= MAX_NETWORK_CLIENTS { - eprintln!("runa-server: rejecting {peer_address}; client limit reached"); - continue; - } - stream.set_nonblocking(true)?; - clients.push(NetworkClient::new(stream, peer_address)); - } - Err(error) if error.kind() == io::ErrorKind::WouldBlock => break, - Err(error) if error.kind() == io::ErrorKind::Interrupted => continue, - Err(error) => return Err(error), - } - } - - let mut index = 0; - while index < clients.len() { - match clients[index].poll(&mut server) { - Ok(true) => index += 1, - Ok(false) => { - clients.swap_remove(index); - } - Err(error) => { - eprintln!("runa-server: client {}: {error}", clients[index].address); - clients.swap_remove(index); - } - } - } - - thread::sleep(Duration::from_millis(5)); - } -} - -struct NetworkClient { - stream: TcpStream, - address: SocketAddr, - input: Vec, - output: Vec, - output_offset: usize, - read_closed: bool, -} - -impl NetworkClient { - fn new(stream: TcpStream, address: SocketAddr) -> Self { - Self { - stream, - address, - input: Vec::new(), - output: Vec::new(), - output_offset: 0, - read_closed: false, - } - } - - fn poll(&mut self, server: &mut Server) -> io::Result { - if self.output_offset == self.output.len() { - self.output.clear(); - self.output_offset = 0; - self.process_pending_requests(server, self.read_closed)?; - - if self.output.is_empty() && !self.read_closed { - let mut buffer = [0; 8192]; - match self.stream.read(&mut buffer) { - Ok(0) => { - self.read_closed = true; - self.process_pending_requests(server, true)?; - } - Ok(length) => { - self.input.extend_from_slice(&buffer[..length]); - self.process_pending_requests(server, false)?; - } - Err(error) if error.kind() == io::ErrorKind::WouldBlock => {} - Err(error) if error.kind() == io::ErrorKind::Interrupted => {} - Err(error) => return Err(error), - } - } - } - - if self.output_offset < self.output.len() { - match self.stream.write(&self.output[self.output_offset..]) { - Ok(0) => { - return Err(io::Error::new( - io::ErrorKind::WriteZero, - "failed to write protocol response", - )); - } - Ok(length) => { - self.output_offset += length; - if self.output_offset == self.output.len() { - self.output.clear(); - self.output_offset = 0; - } - } - Err(error) if error.kind() == io::ErrorKind::WouldBlock => {} - Err(error) if error.kind() == io::ErrorKind::Interrupted => {} - Err(error) => return Err(error), - } - } - - Ok(!self.read_closed || !self.input.is_empty() || self.output_offset < self.output.len()) - } - - fn process_pending_requests( - &mut self, - server: &mut Server, - final_line: bool, - ) -> io::Result<()> { - let Some(line_end) = self.input.iter().position(|byte| *byte == b'\n') else { - if self.input.len() > MAX_REQUEST_LINE_BYTES { - return Err(invalid_data(format!( - "request line exceeds {MAX_REQUEST_LINE_BYTES} bytes" - ))); - } - if !final_line || self.input.is_empty() { - return Ok(()); - } - let line = self.input.clone(); - self.input.clear(); - self.dispatch_line(server, &line)?; - return Ok(()); - }; - - if line_end > MAX_REQUEST_LINE_BYTES { - return Err(invalid_data(format!( - "request line exceeds {MAX_REQUEST_LINE_BYTES} bytes" - ))); - } - let line = self.input[..line_end].to_vec(); - self.input.drain(..=line_end); - self.dispatch_line(server, &line) - } - - fn dispatch_line(&mut self, server: &mut Server, line: &[u8]) -> io::Result<()> { - let request: Request = serde_json::from_slice(line) - .map_err(|error| invalid_data(format!("invalid protocol request JSON: {error}")))?; - let response = server.dispatch(request); - let mut encoded = serde_json::to_vec(&response).map_err(io::Error::other)?; - if encoded.len() > MAX_REQUEST_LINE_BYTES { - return Err(invalid_data(format!( - "response exceeds {MAX_REQUEST_LINE_BYTES} bytes" - ))); - } - encoded.push(b'\n'); - self.output.extend_from_slice(&encoded); - Ok(()) - } -} - fn invalid_data(message: String) -> io::Error { io::Error::new(io::ErrorKind::InvalidData, message) } @@ -555,8 +390,7 @@ mod tests { } use runa_protocol::{Command, PROTOCOL_VERSION}; - use std::io::{BufRead, BufReader, Cursor}; - use std::net::Shutdown; + use std::io::Cursor; fn request(command: Command) -> Request { Request { @@ -773,39 +607,17 @@ mod tests { } #[test] - fn tcp_transport_dispatches_requests_to_the_shared_server() { - let listener = bind_loopback_listener("127.0.0.1:0".parse().unwrap()).unwrap(); - let address = listener.local_addr().unwrap(); - let mut client = TcpStream::connect(address).unwrap(); - let (server_stream, peer_address) = listener.accept().unwrap(); - let mut network_client = NetworkClient::new(server_stream, peer_address); - - let first_request = request(Command::ListActors); - let mut second_request = request(Command::ListActors); - second_request.request_id = "second-request".to_owned(); - writeln!( - client, - "{}\n{}", - serde_json::to_string(&first_request).unwrap(), - serde_json::to_string(&second_request).unwrap() - ) - .unwrap(); - client.shutdown(Shutdown::Write).unwrap(); - - assert!(network_client.poll(&mut Server::new()).unwrap()); - let mut response_line = String::new(); - let mut client_reader = BufReader::new(client); - client_reader.read_line(&mut response_line).unwrap(); - let first_response: Response = serde_json::from_str(&response_line).unwrap(); - assert_eq!(first_response.request_id, first_request.request_id); - - assert!(network_client.poll(&mut Server::new()).unwrap()); - response_line.clear(); - client_reader.read_line(&mut response_line).unwrap(); - let second_response: Response = serde_json::from_str(&response_line).unwrap(); - assert_eq!(second_response.request_id, second_request.request_id); - assert_eq!(second_response.result, Ok(Reply::Actors(Vec::new()))); - - assert!(!network_client.poll(&mut Server::new()).unwrap()); + fn notification_contract_is_versioned_and_distinct_from_responses() { + let notification = runa_protocol::Notification { + version: PROTOCOL_VERSION, + event: runa_protocol::ImageEvent::ImageChanged { revision: 42 }, + }; + assert_eq!( + serde_json::to_value(notification).unwrap(), + serde_json::json!({ + "version": 3, + "event": {"event": "image_changed", "data": {"revision": 42}} + }) + ); } } diff --git a/crates/runa-server/src/main.rs b/crates/runa-server/src/main.rs index aee5be3..2699c66 100644 --- a/crates/runa-server/src/main.rs +++ b/crates/runa-server/src/main.rs @@ -3,7 +3,7 @@ use std::net::SocketAddr; use std::process::ExitCode; use runa_protocol::{Command, PROTOCOL_VERSION, Request}; -use runa_server::{Server, bind_loopback_listener, serve_json_lines, serve_tcp}; +use runa_server::{Server, bind_loopback_listener, serve_json_lines, serve_websocket}; fn initial_image() -> io::Result { let mut server = Server::new(); @@ -55,7 +55,7 @@ fn run() -> io::Result<()> { format!("invalid listener address: {error}"), ) })?; - serve_tcp(bind_loopback_listener(address)?, initial_image()?) + serve_websocket(bind_loopback_listener(address)?, initial_image()?) } Some(_) => Err(io::Error::new( io::ErrorKind::InvalidInput, diff --git a/crates/runa-server/src/network.rs b/crates/runa-server/src/network.rs new file mode 100644 index 0000000..fc6157b --- /dev/null +++ b/crates/runa-server/src/network.rs @@ -0,0 +1,175 @@ +use std::io; +use std::net::TcpListener; +use std::time::Duration; + +use futures_util::{SinkExt, StreamExt}; +use runa_protocol::{ + Command, ImageEvent, Notification, PROTOCOL_VERSION, Request, Response, WEBSOCKET_SUBPROTOCOL, +}; +use tokio::sync::{mpsc, oneshot, watch}; +use tokio::task::JoinSet; +use tokio::time::timeout; +use tokio_tungstenite::tungstenite::handshake::server::{ + Callback, ErrorResponse, Request as UpgradeRequest, Response as UpgradeResponse, +}; +use tokio_tungstenite::tungstenite::http::{HeaderValue, StatusCode}; +use tokio_tungstenite::tungstenite::{Message, protocol::WebSocketConfig}; + +use crate::{MAX_REQUEST_LINE_BYTES, Server}; + +const MAX_NETWORK_CLIENTS: usize = 16; +const WRITE_TIMEOUT: Duration = Duration::from_secs(30); + +/// Keep all image dispatch on one thread, with bounded per-client I/O tasks. +pub fn serve_websocket(listener: TcpListener, server: Server) -> io::Result<()> { + if !listener.local_addr()?.ip().is_loopback() { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + "network listener must be bound to a loopback address", + )); + } + listener.set_nonblocking(true)?; + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build()? + .block_on(serve(listener, server)) +} + +struct Dispatch { + request: Request, + respond: oneshot::Sender, +} + +async fn serve(listener: TcpListener, mut server: Server) -> io::Result<()> { + let listener = tokio::net::TcpListener::from_std(listener)?; + eprintln!("runa-server: listening on {}", listener.local_addr()?); + let (requests, mut pending) = mpsc::channel::(MAX_NETWORK_CLIENTS); + let (changes, _) = watch::channel(0_u64); + let mut revision = 0_u64; + let mut clients = JoinSet::new(); + loop { + tokio::select! { + accepted = listener.accept() => { + let (stream, address) = accepted?; + if clients.len() >= MAX_NETWORK_CLIENTS { + eprintln!("runa-server: rejecting {address}; client limit reached"); + continue; + } + let requests = requests.clone(); + let changes = changes.subscribe(); + clients.spawn(async move { + if let Err(error) = client(stream, requests, changes).await { + eprintln!("runa-server: client {address}: {error}"); + } + }); + } + Some(dispatch) = pending.recv() => { + let mutates = !matches!(dispatch.request.command, + Command::GetBehavior { .. } | Command::GetBehaviors | + Command::GetBehaviorSource { .. } | Command::InspectActor { .. } | + Command::ListActors | Command::GetEvents); + let response = server.dispatch(dispatch.request); + let changed = mutates && response.result.is_ok(); + // Dispatch may commit even if the initiating client has disconnected. + let _ = dispatch.respond.send(response); + if changed { + revision = revision.checked_add(1) + .ok_or_else(|| io::Error::other("image notification revisions exhausted"))?; + changes.send_replace(revision); + } + } + Some(result) = clients.join_next(), if !clients.is_empty() => { + result.map_err(io::Error::other)?; + } + } + } +} + +struct DebugProtocol; + +impl Callback for DebugProtocol { + fn on_request( + self, + request: &UpgradeRequest, + mut response: UpgradeResponse, + ) -> Result { + let offered = request + .headers() + .get("sec-websocket-protocol") + .and_then(|value| value.to_str().ok()) + .is_some_and(|value| { + value + .split(',') + .any(|item| item.trim() == WEBSOCKET_SUBPROTOCOL) + }); + // No browser-origin access is enabled for this unauthenticated local API. + if !offered || request.headers().contains_key("origin") || request.uri().path() != "/" { + let mut error = ErrorResponse::new(Some( + "requires runa-debug subprotocol, root path, and a non-browser local client" + .to_owned(), + )); + *error.status_mut() = StatusCode::BAD_REQUEST; + return Err(error); + } + response.headers_mut().insert( + "sec-websocket-protocol", + HeaderValue::from_static(WEBSOCKET_SUBPROTOCOL), + ); + Ok(response) + } +} + +async fn client( + stream: tokio::net::TcpStream, + requests: mpsc::Sender, + mut changes: watch::Receiver, +) -> Result<(), Box> { + let config = WebSocketConfig::default() + .max_message_size(Some(MAX_REQUEST_LINE_BYTES)) + .max_frame_size(Some(MAX_REQUEST_LINE_BYTES)) + .max_write_buffer_size(2 * MAX_REQUEST_LINE_BYTES); + let mut socket = timeout( + Duration::from_secs(5), + tokio_tungstenite::accept_hdr_async_with_config(stream, DebugProtocol, Some(config)), + ) + .await??; + loop { + tokio::select! { + message = socket.next() => { + match message { + Some(Ok(Message::Text(text))) => { + let request: Request = serde_json::from_str(&text)?; + let (respond, response) = oneshot::channel(); + requests.send(Dispatch { request, respond }).await?; + let encoded = serde_json::to_string(&response.await?)?; + if encoded.len() > MAX_REQUEST_LINE_BYTES { + return Err("WebSocket response exceeds 1 MiB".into()); + } + timeout(WRITE_TIMEOUT, socket.send(Message::Text(encoded.into()))).await??; + } + Some(Ok(Message::Ping(_))) => { + timeout(WRITE_TIMEOUT, socket.flush()).await??; + } + Some(Ok(Message::Pong(_))) => {} + Some(Ok(Message::Close(_))) | None => { + timeout(WRITE_TIMEOUT, socket.flush()).await??; + return Ok(()); + } + Some(Ok(_)) => return Err("only text protocol messages are accepted".into()), + Some(Err(error)) => return Err(error.into()), + } + } + changed = changes.changed() => { + changed?; + // Retain only the latest invalidation for each slow client, not an event log. + let revision = *changes.borrow_and_update(); + let notification = Notification { + version: PROTOCOL_VERSION, + event: ImageEvent::ImageChanged { revision }, + }; + let text = serde_json::to_string(¬ification)?; + timeout(WRITE_TIMEOUT, socket.send(Message::Text(text.into()))).await??; + } + } + } +} diff --git a/crates/runa-server/tests/local_client.rs b/crates/runa-server/tests/local_client.rs index f5558da..82e2aee 100644 --- a/crates/runa-server/tests/local_client.rs +++ b/crates/runa-server/tests/local_client.rs @@ -1,11 +1,211 @@ use runa_protocol::Value; -use std::io::{BufRead, BufReader, Write}; -use std::net::{Shutdown, SocketAddr, TcpStream}; +use std::io::{BufRead, BufReader}; +use std::net::SocketAddr; use std::process::{Child, Command as ProcessCommand, Stdio}; -use std::time::Duration; -use runa_client::{LocalClient, TcpClient}; -use runa_protocol::{Command, PROTOCOL_VERSION, Reply, Request, Response}; +use runa_client::{LocalClient, WebSocketClient}; +use runa_protocol::{Command, Reply}; +use tokio_tungstenite::tungstenite::{self, Message, client::IntoClientRequest}; + +fn start_server() -> (ServerProcess, SocketAddr) { + let child = ProcessCommand::new(env!("CARGO_BIN_EXE_runa-server")) + .args(["--listen", "127.0.0.1:0"]) + .stdout(Stdio::null()) + .stderr(Stdio::piped()) + .spawn() + .unwrap(); + let mut server = ServerProcess(child); + let mut stderr = BufReader::new(server.0.stderr.take().unwrap()); + let mut startup = String::new(); + stderr.read_line(&mut startup).unwrap(); + let address = startup + .strip_prefix("runa-server: listening on ") + .expect("server should report its bound address") + .trim() + .parse() + .unwrap(); + std::thread::spawn(move || { + std::io::copy(&mut stderr, &mut std::io::sink()).unwrap(); + }); + (server, address) +} + +fn watcher(address: SocketAddr) -> tungstenite::WebSocket { + let mut request = format!("ws://{address}/").into_client_request().unwrap(); + request + .headers_mut() + .insert("sec-websocket-protocol", "runa-debug".parse().unwrap()); + let stream = std::net::TcpStream::connect(address).unwrap(); + stream + .set_read_timeout(Some(std::time::Duration::from_secs(2))) + .unwrap(); + let (socket, response) = tungstenite::client(request, stream).unwrap(); + assert_eq!(response.headers()["sec-websocket-protocol"], "runa-debug"); + socket +} + +fn revision(socket: &mut tungstenite::WebSocket) -> u64 { + let message = socket.read().unwrap(); + let event: runa_protocol::Notification = + serde_json::from_str(message.to_text().unwrap()).unwrap(); + assert_eq!(event.version, runa_protocol::PROTOCOL_VERSION); + let runa_protocol::ImageEvent::ImageChanged { revision } = event.event; + revision +} + +#[test] +fn websocket_broadcasts_non_destructive_invalidations_for_mailbox_state_and_lifecycle() { + let (_server, address) = start_server(); + let mut first = watcher(address); + let mut second = watcher(address); + let mut client = WebSocketClient::connect(address).unwrap(); + let actor_ref = match client + .request(Command::Spawn { + behavior_name: "counter".to_owned(), + state: Value::Integer(1), + mailbox_capacity: 4, + }) + .unwrap() + .result + .unwrap() + { + Reply::ActorSpawned { actor_ref } => actor_ref, + other => panic!("unexpected spawn: {other:?}"), + }; + let mut previous = 0; + for command in [ + Command::ListActors, + Command::Send { + actor_ref: actor_ref.clone(), + message: Value::Record(std::collections::BTreeMap::from([ + ("op".to_owned(), Value::String("add".to_owned())), + ("value".to_owned(), Value::Integer(3)), + ])), + }, + Command::Tick { + ticks: 100, + instruction_budget: 10, + }, + Command::PauseActor { + actor_ref: actor_ref.clone(), + }, + Command::ResumeActor { + actor_ref: actor_ref.clone(), + }, + Command::StopActor { + actor_ref: actor_ref.clone(), + }, + Command::RestartActor { + actor_ref: actor_ref.clone(), + }, + ] { + if !matches!(command, Command::ListActors) { + client.request(command).unwrap().result.unwrap(); + } + let current = revision(&mut first); + assert!(current > previous); + assert_eq!(revision(&mut second), current); + previous = current; + } + // Both subscribers have seen notifications, but neither consumed runtime events. + let Reply::Events(events) = client.request(Command::GetEvents).unwrap().result.unwrap() else { + panic!("expected runtime events") + }; + assert!( + events + .iter() + .any(|event| matches!(event, runa_protocol::EventDto::Spawned { .. })) + ); + let Reply::Actor(snapshot) = client + .request(Command::InspectActor { actor_ref }) + .unwrap() + .result + .unwrap() + else { + panic!("expected actor") + }; + assert_eq!(snapshot.state, Value::Integer(4)); + + // A read after a failed mutation is a fence: no invalidation should precede its reply. + assert!( + client + .request(Command::Spawn { + behavior_name: "missing".to_owned(), + state: Value::Null, + mailbox_capacity: 1, + }) + .unwrap() + .result + .is_err() + ); + first + .send(Message::Text( + serde_json::to_string(&runa_protocol::Request { + version: 3, + request_id: "read-fence".to_owned(), + command: Command::ListActors, + }) + .unwrap() + .into(), + )) + .unwrap(); + let message = first.read().unwrap(); + let response: runa_protocol::Response = + serde_json::from_str(message.to_text().unwrap()).unwrap(); + assert_eq!(response.request_id, "read-fence"); +} + +#[test] +fn websocket_requires_subprotocol_and_rejects_browser_origins_without_blocking_other_clients() { + let (_server, address) = start_server(); + // An idle HTTP handshake cannot stall the image or another client's upgrade. + let _idle = std::net::TcpStream::connect(address).unwrap(); + for (protocol, origin) in [ + (None, false), + (Some("debug"), false), + (Some("runa-debug"), true), + ] { + let mut request = format!("ws://{address}/").into_client_request().unwrap(); + if let Some(protocol) = protocol { + request + .headers_mut() + .insert("sec-websocket-protocol", protocol.parse().unwrap()); + } + if origin { + request + .headers_mut() + .insert("origin", "https://example.test".parse().unwrap()); + } + let stream = std::net::TcpStream::connect(address).unwrap(); + stream + .set_read_timeout(Some(std::time::Duration::from_secs(2))) + .unwrap(); + assert!(tungstenite::client(request, stream).is_err()); + } + let mut client = WebSocketClient::connect(address).unwrap(); + assert_eq!( + client.request(Command::ListActors).unwrap().result.unwrap(), + Reply::Actors(Vec::new()) + ); +} + +#[test] +fn websocket_rejects_binary_malformed_and_oversized_requests() { + let (_server, address) = start_server(); + for message in [ + Message::Binary(b"{}".to_vec().into()), + Message::Text("{not-json}".into()), + Message::Text(" ".repeat(runa_server::MAX_REQUEST_LINE_BYTES + 1).into()), + ] { + let mut socket = watcher(address); + // An oversized frame may be rejected as soon as its header arrives, + // before the sender finishes writing its payload. + let _ = socket.send(message); + assert!(socket.read().is_err()); + } + let mut client = WebSocketClient::connect(address).unwrap(); + assert!(client.request(Command::ListActors).unwrap().result.is_ok()); +} struct ServerProcess(Child); @@ -65,7 +265,7 @@ fn local_client_launches_server_and_operates_the_same_live_image() { } #[test] -fn tcp_server_accepts_local_protocol_requests() { +fn websocket_server_accepts_local_protocol_requests() { let child = ProcessCommand::new(env!("CARGO_BIN_EXE_runa-server")) .args(["--listen", "127.0.0.1:0"]) .stdout(Stdio::null()) @@ -84,25 +284,12 @@ fn tcp_server_accepts_local_protocol_requests() { .parse::() .unwrap(); - let mut client = TcpStream::connect_timeout(&address, Duration::from_secs(2)).unwrap(); - let request = Request { - version: PROTOCOL_VERSION, - request_id: "tcp-process-test".to_owned(), - command: Command::ListActors, - }; - writeln!(client, "{}", serde_json::to_string(&request).unwrap()).unwrap(); - client.shutdown(Shutdown::Write).unwrap(); - - let mut response_line = String::new(); - BufReader::new(client) - .read_line(&mut response_line) - .unwrap(); - let response: Response = serde_json::from_str(&response_line).unwrap(); - assert_eq!(response.request_id, "tcp-process-test"); + let mut client = WebSocketClient::connect(address).unwrap(); + let response = client.request(Command::ListActors).unwrap(); assert_eq!(response.result, Ok(Reply::Actors(Vec::new()))); - let mut first = TcpClient::connect(address).unwrap(); - let mut second = TcpClient::connect(address).unwrap(); + let mut first = WebSocketClient::connect(address).unwrap(); + let mut second = WebSocketClient::connect(address).unwrap(); let actor_ref = match first .request(Command::Spawn { behavior_name: "counter".to_owned(), @@ -133,7 +320,7 @@ fn tcp_server_accepts_local_protocol_requests() { drop(first); drop(second); - let mut reconnected = TcpClient::connect(address).unwrap(); + let mut reconnected = WebSocketClient::connect(address).unwrap(); assert!(matches!( reconnected .request(Command::GetBehavior { diff --git a/crates/runa/Cargo.toml b/crates/runa/Cargo.toml index df16c00..7fd9898 100644 --- a/crates/runa/Cargo.toml +++ b/crates/runa/Cargo.toml @@ -12,3 +12,4 @@ serde_json = "1" [dev-dependencies] runa-server = { path = "../runa-server" } +tungstenite = "0.28" diff --git a/crates/runa/src/main.rs b/crates/runa/src/main.rs index 1e9b377..63b1077 100644 --- a/crates/runa/src/main.rs +++ b/crates/runa/src/main.rs @@ -5,7 +5,7 @@ use std::io::{self, Write}; use std::net::SocketAddr; use std::num::NonZeroUsize; -use runa_client::TcpClient; +use runa_client::WebSocketClient; use runa_core::ActorRef; use runa_protocol::{Command, Reply, Value}; #[cfg(test)] @@ -27,7 +27,7 @@ trait Connection { fn dispatch(&mut self, command: Command) -> Result; } -impl Connection for TcpClient { +impl Connection for WebSocketClient { fn dispatch(&mut self, command: Command) -> Result { self.request(command) .map_err(|error| error.to_string())? @@ -192,7 +192,7 @@ impl Helper for ReplHelper {} fn main() -> Result<(), Box> { let address = parse_address(&std::env::args().skip(1).collect::>())?; - let mut server = TcpClient::connect(address).map_err(|error| { + let mut server = WebSocketClient::connect(address).map_err(|error| { format!("cannot connect to {address}: {error}; start `mise run serve-network` first") })?; @@ -910,23 +910,47 @@ mod tests { } #[test] - fn tcp_repl_reconnects_to_the_same_image() -> Result<(), Box> { + fn websocket_repl_reconnects_to_the_same_image() -> Result<(), Box> { let listener = std::net::TcpListener::bind("127.0.0.1:0")?; let address = listener.local_addr()?; let worker = std::thread::spawn(move || { let mut image = server(); for _ in 0..2 { let (stream, _) = listener.accept().unwrap(); - runa_server::serve_json_lines( - &mut image, - io::BufReader::new(stream.try_clone().unwrap()), - stream, - ) - .unwrap(); + struct DebugProtocol; + impl tungstenite::handshake::server::Callback for DebugProtocol { + fn on_request( + self, + _: &tungstenite::handshake::server::Request, + mut response: tungstenite::handshake::server::Response, + ) -> Result< + tungstenite::handshake::server::Response, + tungstenite::handshake::server::ErrorResponse, + > { + response.headers_mut().insert( + "sec-websocket-protocol", + runa_protocol::WEBSOCKET_SUBPROTOCOL.parse().unwrap(), + ); + Ok(response) + } + } + let mut socket = tungstenite::accept_hdr(stream, DebugProtocol).unwrap(); + while let Ok(message) = socket.read() { + let Some(text) = message.to_text().ok() else { + break; + }; + let request = serde_json::from_str(text).unwrap(); + let response = image.dispatch(request); + socket + .send(tungstenite::Message::Text( + serde_json::to_string(&response).unwrap().into(), + )) + .unwrap(); + } } }); - let mut client = TcpClient::connect(address)?; + let mut client = WebSocketClient::connect(address)?; let mut output = Vec::new(); run_repl( io::Cursor::new( @@ -943,7 +967,7 @@ mod tests { assert_eq!(actor_ref.len(), 21); drop(client); - let mut client = TcpClient::connect(address)?; + let mut client = WebSocketClient::connect(address)?; let mut output = Vec::new(); run_repl( io::Cursor::new(format!("actors\ninspect @{actor_ref}\nquit\n")), @@ -965,7 +989,7 @@ mod tests { assert_eq!(snapshot.state, Value::Integer(12)); assert_eq!(snapshot.status, runa_protocol::ActorStatusDto::Idle); drop(client); - worker.join().expect("TCP image worker completed"); + worker.join().expect("WebSocket image worker completed"); Ok(()) } diff --git a/docs/architecture.md b/docs/architecture.md index f31f709..6e7aeed 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -63,7 +63,7 @@ Rust supplies a minimal, reliable substrate; increasingly much of the actor fram +-----------------------------------------------------------+ ``` -The graphical world should use internal service calls to the image API. The VS Code extension connects to the same live image through a versioned network transport. CLI tools, tests, and local automation can use in-process calls or a child-process protocol over stdio. The initial TCP transport is unauthenticated and must bind only to loopback; a remote-capable network boundary must authenticate and authorize clients. Neither transport creates a separate actor runtime or image. +The graphical world should use internal service calls to the image API. The VS Code extension connects to the same live image through a versioned network transport. CLI tools, tests, and local automation can use in-process calls or a child-process protocol over stdio. The initial WebSocket transport is unauthenticated and must bind only to loopback; a remote-capable network boundary must authenticate and authorize clients. Neither transport creates a separate actor runtime or image. ## 4. Fundamental Actor Model @@ -235,19 +235,21 @@ These are desired capabilities, not a requirement to implement every endpoint im ### 8.1 Security -Normal actors should not automatically receive development privileges. Arbitrary evaluation, state mutation, code replacement, and process control require explicit capabilities. The stdio bridge is intended for a trusted parent process. The initial TCP bridge accepts unauthenticated connections only on IPv4 or IPv6 loopback and grants each connection the full development API; any local process may connect. It is suitable only for trusted local development. Never bind this endpoint to a non-loopback address or expose it remotely until authentication and authorization are implemented. +Normal actors should not automatically receive development privileges. Arbitrary evaluation, state mutation, code replacement, and process control require explicit capabilities. The stdio bridge is intended for a trusted parent process. The initial WebSocket bridge accepts unauthenticated connections only on IPv4 or IPv6 loopback and grants each connection the full development API; any local process may connect. It is suitable only for trusted local development. Never bind this endpoint to a non-loopback address or expose it remotely until authentication and authorization are implemented. ## 9. VS Code Extension -VS Code is an external client of the same live image, not a separate runtime. The initial development transport is versioned JSON Lines over TCP, bound to loopback and unauthenticated; the extension must connect only to a trusted local image. Remote connections remain out of scope until authentication and authorization are implemented. The extension does not host actors or create a separate image. The local stdio protocol remains useful for CLI tools, tests, and other child-process clients. +VS Code is an external client of the same live image, not a separate runtime. The initial development transport is versioned JSON over runa-debug WebSockets, bound to loopback and unauthenticated; the extension must connect only to a trusted local image. Remote connections remain out of scope until authentication and authorization are implemented. The extension does not host actors or create a separate image. The local stdio protocol remains useful for CLI tools, tests, and other child-process clients. The first extension increment implements explicit connect/disconnect, an actor -explorer, and read-only state/mailbox/execution snapshots. Refresh and inspection -fetch current data on demand; no scheduler ticks or event drains are performed. +explorer, and read-only state/mailbox/execution snapshots. Server `image_changed` +notifications automatically refresh the tree and open documents without +scheduler ticks or event drains. Revisions and refreshes are coalesced rather +than replayed, and changes during a fetch trigger a follow-up refresh. Cached open snapshots are marked disconnected on connection loss and isolated by connection session so reconnecting to another image cannot overwrite them. -The remaining UX below is planned; non-destructive event subscription semantics -still need design. +The remaining UX below is planned; detailed runtime event subscriptions and +durable replay still need design. Required extension UX: @@ -325,8 +327,8 @@ The diagram is conceptual; Cargo dependencies should be kept minimal. A dependen | `runa-runtime` | Scheduling, actor execution coordination, state commits, crash recovery, binding actor semantics to Rune execution | `runa-core`, `runa-actor`, `runa-code`, `runa-rune` | | `runa-image-api` | Trusted in-process application services for spawning, inspecting, sending, editing, and upgrading | `runa-core`, `runa-actor`, `runa-code`, `runa-runtime` | | `runa-protocol` | Versioned transport-neutral request/response/event DTOs and validation | `serde` | -| `runa-server` | Protocol dispatch, JSON-lines stdio, and loopback-only TCP transport | `runa-protocol`, `runa-image-api` | -| `runa-client` | Local child-process client for the JSON-lines stdio protocol | `runa-protocol` | +| `runa-server` | Protocol dispatch, JSON-lines stdio, and loopback-only WebSocket transport | `runa-protocol`, `runa-image-api` | +| `runa-client` | Local child-process stdio and loopback runa-debug WebSocket clients | `runa-protocol` | | `runa-graphics` | Native window, rendering and input mechanisms; no actor policy | `runa-core` if necessary | | `runa-world` | Morphic-style graphical composition and in-world development UI | `runa-image-api`, `runa-graphics` | | `runa` | Startup, configuration, resource wiring, shutdown | Top-level crates it instantiates | @@ -346,8 +348,8 @@ defines versioned, serializable request, response, and event DTOs without depending on a transport or the runtime implementation. `runa-server` now dispatches those commands through the image API and maps service results into protocol replies and typed errors. The dispatcher is exposed through local -child-process JSON-lines stdio and an unauthenticated TCP listener restricted -to IPv4 or IPv6 loopback; both enforce a 1 MiB request-line limit. The TCP +child-process JSON-lines stdio and an unauthenticated WebSocket listener restricted +to IPv4 or IPv6 loopback; both enforce a 1 MiB request limit. The WebSocket listener shares one image across at most 16 concurrent clients. Every local client receives full development API access, so the endpoint is for trusted local development only. In-process consumers call `runa-image-api`, and @@ -464,11 +466,11 @@ mapping, and local-variable inspection remain future work. The first Phase 5 increments are a trusted in-process application API used by the CLI, a transport-neutral versioned protocol contract, and a dispatcher exposed through bounded local JSON-lines stdio. The interactive CLI now uses -`runa-client` to connect over loopback TCP to a server-owned image, rather than +`runa-client` to connect over loopback WebSockets to a server-owned image, rather than embedding a runtime. Multiple clients share actors and behavior revisions; client disconnect/reconnect leaves the running image intact. `runa-client` also launches the stdio server for external local tools. The server provides an -unauthenticated TCP JSON-lines endpoint restricted to loopback for local VS +unauthenticated runa-debug WebSocket endpoint restricted to loopback for local VS Code development. Remote access will require authentication and authorization; neither network transport nor client location changes the single-image actor model. @@ -485,7 +487,9 @@ single-image actor model. Connection management, the actor explorer, and read-only snapshot inspection are implemented in `vscode/`. Spawn/send and lifecycle controls, behavior editing, -diagnostics, hot upgrades, and event subscriptions remain planned. +diagnostics, hot upgrades, and detailed runtime event subscriptions remain planned. +Non-destructive image-change notifications and automatic snapshot refresh are +implemented over the shared `runa-debug` WebSocket transport. ### Phase 8 — Persistence and advanced debugger diff --git a/docs/protocol.md b/docs/protocol.md index 7d78df8..46d3984 100644 --- a/docs/protocol.md +++ b/docs/protocol.md @@ -3,9 +3,9 @@ `runa-protocol` defines versioned request, response, and event data contracts for clients and the `runa-server` dispatcher. It is independent of transport and runtime implementation. `runa-server` accepts JSON Lines over stdio or, -when started with `--listen`, over TCP bound to an IPv4 or IPv6 loopback -address. The TCP listener is unauthenticated and is intended only for trusted -local development. +when started with `--listen`, as WebSocket text messages bound to an IPv4 or +IPv6 loopback address. The listener is unauthenticated and intended only for +trusted local development. Both the REPL and VS Code use WebSockets. ## Version 3 @@ -75,7 +75,7 @@ client syntax, not wire encodings. Each state/message permits a maximum nesting depth of 32 (root depth 0), 4,096 nodes, and 1 MiB of UTF-8 string/key bytes. Requests are also subject to -the transport's 1 MiB total line limit, including tags and envelope. +the transport's 1 MiB total message limit, including tags and envelope. Unsupported value tags, malformed references, floats, and out-of-range integers fail decoding; well-formed requests exceeding value limits receive `invalid_request`. Invalid Rune transition output fails the actor without @@ -84,38 +84,87 @@ any of its sends. `get_events` drains and returns the events currently accumulated by this single-worker image. It is not a durable subscription or replayable event log. -Both transports limit each request line to 1 MiB before parsing. Malformed -JSON or an oversized line terminates that stream with an error; no protocol +Both transports limit each request to 1 MiB before parsing. Malformed +JSON or an oversized message terminates that connection with an error; no protocol response is emitted when a request cannot be decoded. Stdio EOF terminates the -process, while TCP EOF closes only that client connection. The TCP listener +process, while a WebSocket disconnect closes only that client connection. The listener allows up to 16 simultaneous clients and serializes each dispatch through the -same image. TCP response payloads are capped at 1 MiB, with at most one -response queued per client. Every local client can use the full development +same image. WebSocket response payloads are capped at 1 MiB, with one +in-flight dispatch per client and a bounded shared dispatch queue. Every local client can use the full development API; loopback binding is not authentication or authorization. Remote access must not be enabled until appropriate authentication and authorization exist. The `runa-client` crate launches the stdio server locally and validates -response protocol versions and request IDs. Its `TcpClient` also connects to -loopback endpoints using the same bounded framing and correlation checks. -Connection establishment has a 5-second timeout; reads and writes have -30-second timeouts. A failed TCP request invalidates that client connection. +response protocol versions and request IDs. Its `WebSocketClient` connects to +loopback endpoints using bounded WebSocket messages and correlation checks, +validating interleaved notifications while waiting for responses. +Connection establishment has a 5-second timeout; WebSocket writes and response +waits have 30-second timeouts. A failed network request invalidates that client connection. No automatic retry is performed: loss of a response does not imply that the operation failed to commit. -The interactive REPL uses TCP and never owns the image. Disconnecting clients +The interactive REPL uses WebSockets and never owns the image. Disconnecting clients does not stop actors or reset behavior revisions; reconnecting clients can -retrieve them through `list_actors` and `get_behaviors`. The TCP executable +retrieve them through `list_actors` and `get_behaviors`. The network executable publishes the bundled `counter` behavior on image startup, not on connection. The stdio executable continues to start with an empty image. The TypeScript VS Code client currently issues only `list_actors` and -`inspect_actor`. It enforces loopback endpoints, bounded UTF-8 response lines, +`inspect_actor`. It enforces loopback endpoints, bounded UTF-8 text messages, version/request-ID/reply-kind checks, and portable snapshot value validation. It decodes JSON integers as `bigint` and prints them losslessly, including signed 64-bit state and unsigned 64-bit telemetry. It uses the same 5-second connect and 30-second request timeouts, closes on transport/protocol corruption, and never retries automatically. A well-formed server error is surfaced without -invalidating the stream. Refresh is manual; the extension does not call -`get_events`, because doing so would drain events needed by other clients. +invalidating the stream. Server notifications trigger automatic refreshes of +the actor tree and open snapshot documents. Manual refresh remains available. + +## WebSocket transport and notifications + +Connect to `ws://127.0.0.1:3737/` (or a configured numeric loopback endpoint) +and offer the `runa-debug` WebSocket subprotocol. The server requires and echoes +that subprotocol. Plain TCP JSON Lines is no longer accepted by the network +listener. The request/response contract remains version 3; stdio is unchanged. +Each WebSocket text message contains exactly one JSON request, response, or +notification, without newline framing. Binary messages and malformed requests +terminate the connection. Fragmented text is supported with a 1 MiB aggregate +message limit; individual frames are also limited to 1 MiB. Compression is not +enabled. Upgrade handshakes have a 5-second timeout and writes have a +30-second timeout so stalled peers do not hold client slots forever. + +The endpoint accepts only the root path and rejects requests carrying an +`Origin` header: browser-origin access is not enabled for this unauthenticated +local API. This is not authentication; local non-browser clients retain full +development privileges. + +Every upgraded connection automatically receives non-destructive invalidations: + +```json +{ + "version": 3, + "event": { + "event": "image_changed", + "data": { "revision": 42 } + } +} +``` + +Successful mutating requests (spawn, send, tick, lifecycle controls, +publication/rollback, and installation) increment a server-session revision +and notify all connected clients, including the initiator. Read-only requests, +failed mutations, and `get_events` do not generate invalidations. A tick +invalidates snapshots even when it only advances execution telemetry or causes +an actor failure. A successful send invalidates mailbox snapshots before any +handler runs. Notifications may appear between requests and their replies; +clients must distinguish them from correlated responses. + +Revisions increase within one server session, not across server restarts. +Slow subscribers retain only the latest unsent revision; gaps are expected. +These notifications are hints to re-fetch authoritative data, not individual +runtime events, a durable log, an atomic snapshot revision, or replay. +On connection/reconnection, fetch a full listing first. The VS Code client +coalesces bursts and schedules another refresh if an invalidation arrives +during an in-flight read. It never advances the scheduler or calls `get_events`; +the REPL's explicit `events` command still drains the separate runtime queue. The current validation checks the protocol version, request ID, actor-reference shape, structured-value limits, positive revisions where required, non-empty behavior names, mailbox diff --git a/docs/runtime.md b/docs/runtime.md index d9d4901..a0b746d 100644 --- a/docs/runtime.md +++ b/docs/runtime.md @@ -12,7 +12,7 @@ Quitting the REPL leaves the image running. Reconnecting clients can list and inspect the same actors; local bindings and history do not survive a REPL restart. Server exit still loses the entire image. -The TCP server publishes [counter.rn](../runa-std/actors/counter.rn) as the +The WebSocket server publishes [counter.rn](../runa-std/actors/counter.rn) as the `counter` behavior. Commands include `spawn`, `send`, `tick`, `inspect`, `actors`, `events`, `stop`, `restart`, `pause`, `resume`, `cancel`, `behaviors`, `publish`, `history`, `rollback`, and `install`; enter `help` to see usage. @@ -22,7 +22,10 @@ current revision and a target historical revision. `install` selects a named behavior's current implementation and fails if the actor is not at an idle handler boundary. `spawn` requires a behavior name and initial state, in that order. All commands operate on the same process-local runtime and code store. -The CLI sends versioned protocol requests to the loopback-only TCP listener. +The CLI sends versioned requests to the loopback-only WebSocket listener using +the `runa-debug` subprotocol. VS Code uses the same protocol and automatically +refreshes on non-destructive server change notifications. These notifications +do not tick the image or drain the REPL's explicit `events` queue. The separate stdio endpoint starts an empty image for child-process clients. Example: @@ -78,9 +81,10 @@ also offers references inside a `"$actor"` string. event DTOs; it does not dispatch requests or access runtime state. - `runa-server` dispatches those requests through `runa-image-api` and maps results to protocol replies; its executable accepts newline-delimited JSON - over stdin/stdout or TCP bound to loopback, with a 1 MiB request-line limit. -- `runa-client` provides correlated, bounded stdio and loopback TCP clients. -- `runa` provides an interactive CLI using the TCP client, without an embedded + over stdin/stdout or JSON text messages over loopback WebSockets, with a + 1 MiB request limit. +- `runa-client` provides correlated, bounded stdio and loopback WebSocket clients. +- `runa` provides an interactive CLI using the WebSocket client, without an embedded runtime. Connection errors are explicit; requests are never auto-retried. The image API is process-local and intended for trusted callers. It has no diff --git a/mise.toml b/mise.toml index 7340894..a03904d 100644 --- a/mise.toml +++ b/mise.toml @@ -26,7 +26,7 @@ description = "Run the local JSON-lines protocol server" run = "cargo run -p runa-server" [tasks.serve-network] -description = "Run the protocol server on localhost" +description = "Run the runa-debug WebSocket server on localhost" run = "cargo run -p runa-server -- --listen 127.0.0.1:3737" [tasks.lint] diff --git a/vscode/README.md b/vscode/README.md index cd24f52..481f2f4 100644 --- a/vscode/README.md +++ b/vscode/README.md @@ -1,7 +1,8 @@ # Runa VS Code client This is the first, read-only client for a trusted localhost Runa image. It uses -the [version 3 JSON-lines protocol](../docs/protocol.md), not Rust APIs, and never +the [version 3 protocol](../docs/protocol.md) over WebSockets with the `runa-debug` +subprotocol, not Rust APIs, and never hosts actors or starts its own image. ## Run from source @@ -24,7 +25,7 @@ In that host: 1. Run **Runa: Connect to Local Image**. The default is `127.0.0.1:3737`; `runa.address` changes the suggested endpoint. 2. Expand **Runa Actors** in Explorer. Use the REPL (`mise run start`) to - spawn actors in the same server, then run **Runa: Refresh Actors**. + spawn actors in the same server. The tree updates automatically. 3. Click an actor or run **Runa: Inspect Actor** to open its JSON snapshot. 4. Run **Runa: Disconnect** to close only this client connection. @@ -44,7 +45,9 @@ machine running the extension host, not necessarily the desktop. strings and preserve signed 64-bit integers exactly. - Documents are read-only snapshots, not editable actor state or live VM frames. The timestamp records when this client received the data, not a runtime clock. - **Refresh Actors** updates the tree and existing documents for that connection. + Server `image_changed` notifications automatically update the tree and open + documents. Bursts are coalesced, and changes arriving during a fetch trigger + a follow-up fetch. **Refresh Actors** is also available on demand. - Open documents are retained and marked `connected: false` on disconnect. Each connection gets distinct document URIs; reconnecting never replaces snapshots from an earlier connection or another image. Closing a document @@ -55,6 +58,8 @@ machine running the extension host, not necessarily the desktop. output channel. Server-reported operation errors leave the connection usable. - The client does not tick the scheduler, drain `get_events`, spawn/send, or change actor lifecycle/behavior. Use the REPL for those operations. + Notifications are non-destructive invalidations, not a replayable runtime + event stream; reconnect fetches the current image rather than replaying history. For automation, the `runa.connect` command also accepts an explicit endpoint string, and `runa.inspect` accepts an object with an `actor_ref` field. @@ -71,18 +76,20 @@ mise run test-extension-host ``` The normal test task covers portable-value validation, framing and UTF-8 -fragmentation, timeout/disconnect behavior, exact 64-bit integer display, and +fragmentation, subprotocol negotiation, interleaved notifications, refresh +coalescing, timeout/disconnect behavior, exact 64-bit integer display, and interoperation with the real Rust server on an ephemeral loopback port. The separate host test downloads VS Code 1.96.4 to the ignored `.vscode-test/` -cache and runs connection, inspection, refresh, disconnect, and reconnect -commands against a test-owned Rust server. It verifies actual virtual document -content and session isolation. On headless Linux: +cache and runs connection, inspection, disconnect, and reconnect commands +against a test-owned Rust server. It verifies automatic document updates after +another client sends messages, pauses, and advances an actor into failure, as +well as session isolation. On headless Linux: ```sh xvfb-run -a mise run test-extension-host ``` Behavior/source browsing and editing, spawn/send and lifecycle controls, -compilation diagnostics, non-destructive event subscriptions, packaging for +compilation diagnostics, detailed runtime event subscriptions, packaging for distribution, and debugging remain future work. diff --git a/vscode/integration/host.ts b/vscode/integration/host.ts index 4da671d..15afdf2 100644 --- a/vscode/integration/host.ts +++ b/vscode/integration/host.ts @@ -16,6 +16,9 @@ export async function run(): Promise { assert.match(document.getText(), /9223372036854775807/); assert.match(document.getText(), /"connected": true/); const firstUri = document.uri.toString(); + const refreshed = documentChanged(document, (text) => + text.includes("queued from another client"), + ); assert.deepEqual( await control(image.address, { operation: "send", @@ -26,11 +29,22 @@ export async function run(): Promise { }), { version: 3, request_id: "integration-control", result: { Ok: { kind: "acknowledged" } } }, ); - const refreshed = documentChanged(document); - await vscode.commands.executeCommand("runa.refresh"); await refreshed; assert.match(document.getText(), /queued from another client/); - const disconnected = documentChanged(document); + const paused = documentChanged(document, (text) => text.includes('"status": "paused"')); + await control(image.address, { operation: "pause_actor", parameters: { actor_ref: actorRef } }); + await paused; + const failed = documentChanged(document, (text) => text.includes('"status": "failed"')); + await control(image.address, { + operation: "resume_actor", + parameters: { actor_ref: actorRef }, + }); + await control(image.address, { + operation: "tick", + parameters: { ticks: 100, instruction_budget: 10 }, + }); + await failed; + const disconnected = documentChanged(document, (text) => text.includes('"connected": false')); await vscode.commands.executeCommand("runa.disconnect"); await disconnected; assert.match(document.getText(), /"connected": false/); @@ -42,17 +56,25 @@ export async function run(): Promise { assert.match(document.getText(), /queued from another client/); const old = await vscode.workspace.openTextDocument(vscode.Uri.parse(firstUri)); assert.match(old.getText(), /"connected": false/); - console.log("Runa extension host: connect, inspect, refresh, disconnect, reconnect passed"); + console.log( + "Runa extension host: connect, inspect, automatic refresh, disconnect, reconnect passed", + ); } finally { await vscode.commands.executeCommand("runa.disconnect"); await image.stop(); } } -function documentChanged(document: vscode.TextDocument): Promise { +function documentChanged( + document: vscode.TextDocument, + expected: (text: string) => boolean, +): Promise { return new Promise((resolve, reject) => { const subscription = vscode.workspace.onDidChangeTextDocument((event) => { - if (event.document.uri.toString() === document.uri.toString()) { + if ( + event.document.uri.toString() === document.uri.toString() && + expected(event.document.getText()) + ) { clearTimeout(timeout); subscription.dispose(); resolve(); diff --git a/vscode/integration/repl.test.ts b/vscode/integration/repl.test.ts new file mode 100644 index 0000000..f02b382 --- /dev/null +++ b/vscode/integration/repl.test.ts @@ -0,0 +1,73 @@ +import assert from "node:assert/strict"; +import { spawn } from "node:child_process"; +import { once } from "node:events"; +import { resolve } from "node:path"; +import test from "node:test"; +import { ImageClient } from "../src/client"; +import { startImage } from "./server"; + +test( + "the real REPL mutates the same WebSocket image and notifies the inspection client", + { timeout: 10000 }, + async (t) => { + const image = await startImage(); + let inspector: ImageClient | undefined; + t.after(async () => { + inspector?.close(); + await image.stop(); + }); + const revisions: bigint[] = []; + let onChanged = () => {}; + const changed = new Promise((resolve) => { + onChanged = resolve; + }); + inspector = await ImageClient.connect( + image.address, + (error) => assert.equal(error, undefined), + (revision) => { + revisions.push(revision); + onChanged(); + }, + ); + const executable = resolve( + __dirname, + "../../../target/debug", + process.platform === "win32" ? "runa.exe" : "runa", + ); + const repl = spawn(executable, ["--connect", image.address], { + stdio: ["pipe", "pipe", "pipe"], + }); + t.after(() => { + if (repl.exitCode === null && repl.signalCode === null) repl.kill(); + }); + let stdout = ""; + let stderr = ""; + repl.stdout.on("data", (chunk: Buffer) => { + stdout += chunk.toString(); + }); + repl.stderr.on("data", (chunk: Buffer) => { + stderr += chunk.toString(); + }); + const exited = once(repl, "exit"); + repl.stdin.end( + [ + "let actor = spawn counter 5", + 'send actor {"op":"add","value":7}', + "tick 100", + "inspect actor", + "quit", + ].join("\n") + "\n", + ); + const [code] = await exited; + assert.equal(code, 0, stderr); + assert.doesNotMatch(stdout, /error:/i); + await changed; + assert.ok(revisions.length > 0); + const [actor] = await inspector.listActors(); + assert.ok(actor); + assert.ok(stdout.includes(actor.actor_ref)); + assert.deepEqual(actor.state, { type: "integer", data: 12n }); + assert.deepEqual(actor.queued, []); + assert.equal(actor.reserved, null); + }, +); diff --git a/vscode/integration/server.test.ts b/vscode/integration/server.test.ts index 9dbbf30..73fffdc 100644 --- a/vscode/integration/server.test.ts +++ b/vscode/integration/server.test.ts @@ -3,50 +3,76 @@ import test from "node:test"; import { ImageClient } from "../src/client"; import { control, spawnCounter, startImage } from "./server"; -test("TypeScript inspection shares the real Rust image across clients and reconnects", async (t) => { - const image = await startImage(); - t.after(() => image.stop()); - const actorRef = await spawnCounter(image.address); - const reference = { type: "actor_ref", data: actorRef }; - const payload = { - type: "record", - data: { - nested: { - type: "list", - data: [reference, { type: "integer", data: -9223372036854775808n }], +test( + "TypeScript inspection shares the real Rust image across clients and reconnects", + { timeout: 10000 }, + async (t) => { + const image = await startImage(); + t.after(() => image.stop()); + const actorRef = await spawnCounter(image.address); + const reference = { type: "actor_ref", data: actorRef }; + const payload = { + type: "record", + data: { + nested: { + type: "list", + data: [reference, { type: "integer", data: -9223372036854775808n }], + }, }, - }, - }; - assert.deepEqual( - await control(image.address, { - operation: "send", - parameters: { actor_ref: actorRef, message: payload }, - }), - { version: 3, request_id: "integration-control", result: { Ok: { kind: "acknowledged" } } }, - ); + }; + assert.deepEqual( + await control(image.address, { + operation: "send", + parameters: { actor_ref: actorRef, message: payload }, + }), + { version: 3, request_id: "integration-control", result: { Ok: { kind: "acknowledged" } } }, + ); - const failures: Error[] = []; - const disconnected = (error?: Error) => { - if (error) failures.push(error); - }; - const first = await ImageClient.connect(image.address, disconnected); - t.after(() => first.close()); - const [snapshot] = await first.listActors(); - assert.ok(snapshot); - assert.equal(snapshot.actor_ref, actorRef); - assert.deepEqual(snapshot.state, { type: "integer", data: 9223372036854775807n }); - assert.deepEqual(snapshot.queued, [payload]); - assert.equal(snapshot.reserved, null); - first.close(); - const second = await ImageClient.connect(image.address, disconnected); - t.after(() => second.close()); - assert.deepEqual(await second.inspectActor(actorRef), snapshot); - // Reading from the extension must not drain the shared runtime event queue. - const events = await control(image.address, { operation: "get_events" }); - assert.deepEqual(events, { - version: 3, - request_id: "integration-control", - result: { Ok: { kind: "events", data: [{ event: "spawned", data: { actor_ref: actorRef } }] } }, - }); - assert.deepEqual(failures, []); -}); + const failures: Error[] = []; + const disconnected = (error?: Error) => { + if (error) failures.push(error); + }; + const first = await ImageClient.connect(image.address, disconnected); + t.after(() => first.close()); + const [snapshot] = await first.listActors(); + assert.ok(snapshot); + assert.equal(snapshot.actor_ref, actorRef); + assert.deepEqual(snapshot.state, { type: "integer", data: 9223372036854775807n }); + assert.deepEqual(snapshot.queued, [payload]); + assert.equal(snapshot.reserved, null); + first.close(); + const revisions: bigint[] = []; + let onChanged = () => {}; + const changed = new Promise((resolve) => { + onChanged = resolve; + }); + const second = await ImageClient.connect(image.address, disconnected, (revision) => { + revisions.push(revision); + onChanged(); + }); + t.after(() => second.close()); + assert.deepEqual(await second.inspectActor(actorRef), snapshot); + await control(image.address, { operation: "pause_actor", parameters: { actor_ref: actorRef } }); + await changed; + assert.equal(revisions.length, 1); + assert.ok(revisions[0] !== undefined && revisions[0] > 0n); + const paused = await second.inspectActor(actorRef); + assert.equal(paused.status.status, "paused"); + // Reading from the extension must not drain the shared runtime event queue. + const events = await control(image.address, { operation: "get_events" }); + assert.deepEqual(events, { + version: 3, + request_id: "integration-control", + result: { + Ok: { + kind: "events", + data: [ + { event: "spawned", data: { actor_ref: actorRef } }, + { event: "paused", data: { actor_ref: actorRef } }, + ], + }, + }, + }); + assert.deepEqual(failures, []); + }, +); diff --git a/vscode/integration/server.ts b/vscode/integration/server.ts index 69b085f..3904420 100644 --- a/vscode/integration/server.ts +++ b/vscode/integration/server.ts @@ -1,9 +1,10 @@ import { ChildProcess, spawn } from "node:child_process"; import { once } from "node:events"; -import { createConnection } from "node:net"; import { resolve } from "node:path"; import { createInterface } from "node:readline"; import { json } from "../test/fixtures"; +import WebSocket from "ws"; +import { decodeNotification, WEBSOCKET_SUBPROTOCOL } from "../src/protocol"; export async function startImage(): Promise<{ address: string; stop: () => Promise }> { const executable = resolve( @@ -62,28 +63,30 @@ async function stopImage(child: ChildProcess): Promise { } export async function control(address: string, command: unknown): Promise { - const port = Number(address.split(":")[1]); - const socket = createConnection({ host: "127.0.0.1", port }); - socket.setTimeout(5000, () => socket.destroy(new Error("Control request timed out"))); - const lines = createInterface({ input: socket }); + const socket = new WebSocket(`ws://${address}/`, WEBSOCKET_SUBPROTOCOL); + const timer = setTimeout(() => socket.terminate(), 5000); try { - await once(socket, "connect"); - const response = Promise.race([ - once(lines, "line").then((args) => { - const line: unknown = args[0]; - if (typeof line !== "string") throw new Error("Missing control response"); - const decoded: unknown = JSON.parse(line); - return decoded; - }), - once(socket, "close").then(() => { - throw new Error("Control connection closed"); - }), - ]); - socket.write(json({ version: 3, request_id: "integration-control", command }) + "\n"); + await once(socket, "open"); + const response = new Promise((resolve, reject) => { + socket.on("message", (data, binary) => { + try { + if (binary) throw new Error("Unexpected binary control reply"); + const text = data.toString(); + if (decodeNotification(text) !== undefined) return; + const decoded: unknown = JSON.parse(text); + resolve(decoded); + } catch (error) { + reject(error); + } + }); + socket.once("close", () => reject(new Error("Control connection closed"))); + socket.once("error", reject); + }); + socket.send(json({ version: 3, request_id: "integration-control", command })); return await response; } finally { - lines.close(); - socket.destroy(); + clearTimeout(timer); + socket.close(); } } diff --git a/vscode/integration/websocket.test.ts b/vscode/integration/websocket.test.ts new file mode 100644 index 0000000..9b21758 --- /dev/null +++ b/vscode/integration/websocket.test.ts @@ -0,0 +1,60 @@ +import assert from "node:assert/strict"; +import { once } from "node:events"; +import test from "node:test"; +import WebSocket from "ws"; +import { MAX_MESSAGE_BYTES, WEBSOCKET_SUBPROTOCOL } from "../src/protocol"; +import { json } from "../test/fixtures"; +import { control, startImage } from "./server"; + +test( + "Rust server accepts fragmented text at the exact message limit and handles ping", + { timeout: 10000 }, + async (t) => { + const image = await startImage(); + t.after(() => image.stop()); + const socket = new WebSocket(`ws://${image.address}/`, WEBSOCKET_SUBPROTOCOL); + t.after(() => socket.terminate()); + await once(socket, "open"); + assert.equal(socket.protocol, WEBSOCKET_SUBPROTOCOL); + const request = json({ + version: 3, + request_id: "fragmented", + command: { operation: "list_actors" }, + }); + const text = request + " ".repeat(MAX_MESSAGE_BYTES - Buffer.byteLength(request)); + const reply = once(socket, "message"); + socket.send(text.slice(0, 100), { fin: false }); + socket.send(text.slice(100), { fin: true }); + const [response] = await reply; + assert.deepEqual(JSON.parse(String(response)), { + version: 3, + request_id: "fragmented", + result: { Ok: { kind: "actors", data: [] } }, + }); + const pong = once(socket, "pong"); + socket.ping("runa-debug"); + const [payload] = await pong; + assert.equal(String(payload), "runa-debug"); + }, +); + +test( + "Rust server rejects an oversized fragmented aggregate and keeps the image available", + { timeout: 10000 }, + async (t) => { + const image = await startImage(); + t.after(() => image.stop()); + const socket = new WebSocket(`ws://${image.address}/`, WEBSOCKET_SUBPROTOCOL); + t.after(() => socket.terminate()); + await once(socket, "open"); + const closed = once(socket, "close"); + socket.send(" ".repeat(MAX_MESSAGE_BYTES), { fin: false }); + socket.send(" ", { fin: true }); + await closed; + assert.deepEqual(await control(image.address, { operation: "list_actors" }), { + version: 3, + request_id: "integration-control", + result: { Ok: { kind: "actors", data: [] } }, + }); + }, +); diff --git a/vscode/package-lock.json b/vscode/package-lock.json index 58aa7ff..888c326 100644 --- a/vscode/package-lock.json +++ b/vscode/package-lock.json @@ -8,11 +8,13 @@ "name": "runa", "version": "0.0.0", "dependencies": { - "lossless-json": "^4.3.0" + "lossless-json": "^4.3.0", + "ws": "^8.18.3" }, "devDependencies": { "@types/node": "^22.0.0", "@types/vscode": "~1.96.0", + "@types/ws": "^8.18.1", "@vscode/test-electron": "^2.5.2", "prettier": "^3.6.2", "typescript": "^5.9.3" @@ -39,6 +41,16 @@ "dev": true, "license": "MIT" }, + "node_modules/@types/ws": { + "version": "8.18.2", + "resolved": "https://registry.npmjs.org/@types/ws/-/ws-8.18.2.tgz", + "integrity": "sha512-67MQl+fpWKVTT1NYdnmo3U4sc/xPo/zQBncVnI74qmQa0z/b+1g6iYqNmGCPbxO+zz2aklb08a0oHfegiVd0/w==", + "dev": true, + "license": "MIT", + "dependencies": { + "@types/node": "*" + } + }, "node_modules/@vscode/test-electron": { "version": "2.5.2", "resolved": "https://registry.npmjs.org/@vscode/test-electron/-/test-electron-2.5.2.tgz", @@ -547,6 +559,27 @@ "integrity": "sha512-EPD5q1uXyFxJpCrLnCc1nHnq3gOa6DZBocAIiI2TaSCA7VCJ1UJDMagCzIkXNsUYfD1daK//LTEQ8xiIbrHtcw==", "dev": true, "license": "MIT" + }, + "node_modules/ws": { + "version": "8.22.1", + "resolved": "https://registry.npmjs.org/ws/-/ws-8.22.1.tgz", + "integrity": "sha512-oyL6OPYme/CmiFvNuGzeoxCJ4nOrZE2O1FwM9MWM1AsGri6nLXB2pg+lNA8aLeXMSav0J6eIVS1j6vUPzUtK9g==", + "license": "MIT", + "engines": { + "node": ">=10.0.0" + }, + "peerDependencies": { + "bufferutil": "^4.0.1", + "utf-8-validate": ">=5.0.2" + }, + "peerDependenciesMeta": { + "bufferutil": { + "optional": true + }, + "utf-8-validate": { + "optional": true + } + } } } } diff --git a/vscode/package.json b/vscode/package.json index 4f72697..15c030e 100644 --- a/vscode/package.json +++ b/vscode/package.json @@ -20,16 +20,18 @@ "lint": "tsc -p . --noEmit && prettier --check .", "format": "prettier --write .", "test": "npm run compile && node --test out/test/*.test.js", - "test:integration": "cargo build --manifest-path ../Cargo.toml -p runa-server && npm run compile && node --test out/integration/*.test.js", + "test:integration": "cargo build --manifest-path ../Cargo.toml -p runa-server -p runa && npm run compile && node --test out/integration/*.test.js", "test:host": "cargo build --manifest-path ../Cargo.toml -p runa-server && npm run compile && node scripts/test-host.cjs" }, "dependencies": { - "lossless-json": "^4.3.0" + "lossless-json": "^4.3.0", + "ws": "^8.18.3" }, "devDependencies": { "@vscode/test-electron": "^2.5.2", "@types/node": "^22.0.0", "@types/vscode": "~1.96.0", + "@types/ws": "^8.18.1", "prettier": "^3.6.2", "typescript": "^5.9.3" }, diff --git a/vscode/src/client.ts b/vscode/src/client.ts index d97944b..2ed15dd 100644 --- a/vscode/src/client.ts +++ b/vscode/src/client.ts @@ -1,14 +1,17 @@ -import { createConnection, isIP, Socket } from "node:net"; +import { isIP } from "node:net"; import { TextDecoder } from "node:util"; +import WebSocket, { RawData } from "ws"; import { ActorSnapshot, + decodeNotification, decodeResponse, isActorRef, - MAX_LINE_BYTES, + MAX_MESSAGE_BYTES, PROTOCOL_VERSION, ProtocolError, ReadCommand, ReadReply, + WEBSOCKET_SUBPROTOCOL, } from "./protocol"; export function parseEndpoint(endpoint: string): { host: string; port: number } { @@ -22,7 +25,8 @@ export function parseEndpoint(endpoint: string): { host: string; port: number } throw new Error("Only numeric IPv4 loopback or [::1] endpoints are supported"); } const port = Number(match[3]); - if (!Number.isInteger(port) || port < 1 || port > 65535) throw new Error("Invalid TCP port"); + if (!Number.isInteger(port) || port < 1 || port > 65535) + throw new Error("Invalid WebSocket port"); return { host, port }; } @@ -35,52 +39,61 @@ interface Pending { } export class ImageClient { - private buffer = Buffer.alloc(0); private pending?: Pending; private queue: Promise = Promise.resolve(); private sequence = 0; + private revision = 0n; private closed = false; + private ready = false; private constructor( - private readonly socket: Socket, + private readonly socket: WebSocket, private readonly onDisconnect: (error?: Error) => void, + private readonly onImageChanged: (revision: bigint) => void, private readonly requestTimeout: number, ) { - socket.on("data", (chunk: Buffer) => this.receive(chunk)); + socket.on("message", (data, binary) => this.receive(data, binary)); socket.on("error", (error) => this.close(error)); - socket.on("close", () => this.close(new Error("Image connection closed"))); + socket.on("close", () => this.close(new Error("Image WebSocket closed"))); } static async connect( endpoint: string, onDisconnect: (error?: Error) => void, + onImageChanged: (revision: bigint) => void = () => {}, timeouts = { connect: 5000, request: 30000 }, ): Promise { - const address = parseEndpoint(endpoint); - const socket = createConnection(address); - await new Promise((resolve, reject) => { - const cleanup = () => { - clearTimeout(timer); - socket.off("error", fail); - socket.off("connect", connected); - }; - const fail = (error: Error) => { - cleanup(); - socket.destroy(); - reject(error); - }; - const connected = () => { - cleanup(); - resolve(); - }; - const timer = setTimeout( - () => fail(new Error("Image connection timed out")), - timeouts.connect, - ); - socket.once("error", fail); - socket.once("connect", connected); + parseEndpoint(endpoint); + const socket = new WebSocket(`ws://${endpoint}/`, WEBSOCKET_SUBPROTOCOL, { + handshakeTimeout: timeouts.connect, + maxPayload: MAX_MESSAGE_BYTES, + perMessageDeflate: false, + followRedirects: false, }); - return new ImageClient(socket, onDisconnect, timeouts.request); + const client = new ImageClient(socket, onDisconnect, onImageChanged, timeouts.request); + try { + await new Promise((resolve, reject) => { + const fail = (error: Error) => { + socket.off("open", opened); + reject(error); + }; + const opened = () => { + socket.off("error", fail); + if (socket.protocol !== WEBSOCKET_SUBPROTOCOL) { + reject(new Error("Server did not negotiate runa-debug")); + } else { + client.ready = true; + resolve(); + } + }; + socket.once("error", fail); + socket.once("open", opened); + }); + } catch (error) { + client.close(); + throw error; + } + return client; } async listActors(): Promise { @@ -108,7 +121,7 @@ export class ImageClient { return; } const requestId = `runa-vscode-${++this.sequence}`; - const line = JSON.stringify({ + const message = JSON.stringify({ version: PROTOCOL_VERSION, request_id: requestId, command, @@ -118,12 +131,12 @@ export class ImageClient { this.requestTimeout, ); this.pending = { requestId, command, resolve, reject, timer }; - this.socket.write(line + "\n", (error) => { + this.socket.send(message, (error) => { if (error) this.close(error); }); }), ); - // A failed request must not prevent later requests; each caller retains its own rejection. + // Each caller retains its own rejection, without poisoning subsequent requests. this.queue = request.then( () => undefined, () => undefined, @@ -131,48 +144,49 @@ export class ImageClient { return request; } - private receive(chunk: Buffer): void { + private receive(data: RawData, binary: boolean): void { if (this.closed) return; - let offset = 0; - while (offset < chunk.length && !this.closed) { - if (!this.pending) { - this.close(new Error("Unexpected unsolicited image response")); + try { + if (binary) throw new Error("Only WebSocket text protocol messages are accepted"); + const bytes = Array.isArray(data) + ? Buffer.concat(data) + : data instanceof ArrayBuffer + ? Buffer.from(data) + : data; + if (bytes.length > MAX_MESSAGE_BYTES) throw new Error("Image response exceeds 1 MiB"); + const text = new TextDecoder("utf-8", { fatal: true }).decode(bytes); + const revision = decodeNotification(text); + if (revision !== undefined) { + if (revision <= this.revision) + throw new Error("Non-increasing image notification revision"); + this.revision = revision; + this.onImageChanged(revision); return; } - const newline = chunk.indexOf(10, offset); - const end = newline === -1 ? chunk.length : newline; - if (this.buffer.length + end - offset > MAX_LINE_BYTES) { - this.close(new Error("Image response exceeds the 1 MiB line limit")); - return; - } - this.buffer = Buffer.concat([this.buffer, chunk.subarray(offset, end)]); - if (newline === -1) return; - const frame = this.buffer; - this.buffer = Buffer.alloc(0); - offset = newline + 1; const pending = this.pending; + if (!pending) throw new Error("Unexpected unsolicited image response"); + try { + pending.resolve(decodeResponse(text, pending.requestId, pending.command)); + } catch (error) { + if (!(error instanceof ProtocolError)) throw error; + pending.reject(error); + } this.pending = undefined; clearTimeout(pending.timer); - try { - const line = new TextDecoder("utf-8", { fatal: true }).decode(frame); - pending.resolve(decodeResponse(line, pending.requestId, pending.command)); - } catch (error) { - const failure = error instanceof Error ? error : new Error(String(error)); - pending.reject(failure); - if (!(failure instanceof ProtocolError)) this.close(failure); - } + } catch (error) { + this.close(error instanceof Error ? error : new Error(String(error))); } } close(error?: Error): void { if (this.closed) return; this.closed = true; - this.socket.destroy(); + this.socket.terminate(); if (this.pending) { clearTimeout(this.pending.timer); this.pending.reject(error ?? new Error("Disconnected from image")); this.pending = undefined; } - this.onDisconnect(error); + if (this.ready) this.onDisconnect(error); } } diff --git a/vscode/src/extension.ts b/vscode/src/extension.ts index 462cd2e..8b04bc8 100644 --- a/vscode/src/extension.ts +++ b/vscode/src/extension.ts @@ -1,6 +1,7 @@ import * as vscode from "vscode"; import { ImageClient, parseEndpoint } from "./client"; import { ActorSnapshot, formatSnapshot } from "./protocol"; +import { RefreshQueue } from "./refresh"; class ActorTree implements vscode.TreeDataProvider { private readonly changes = new vscode.EventEmitter(); @@ -123,6 +124,7 @@ export function activate(context: vscode.ExtensionContext): void { let session = 0; let endpoint = ""; let disposed = false; + let automaticRefresh: RefreshQueue | undefined; const report = (error: unknown) => { const failure = error instanceof Error ? error : new Error(String(error)); @@ -144,6 +146,8 @@ export function activate(context: vscode.ExtensionContext): void { }; const disconnect = () => { + automaticRefresh?.dispose(); + automaticRefresh = undefined; const previous = client; client = undefined; documents.disconnect(session); @@ -190,17 +194,27 @@ export function activate(context: vscode.ExtensionContext): void { const currentSession = session; view.message = `Connecting to ${address}...`; let connected: ImageClient; + const updates = new RefreshQueue(async () => { + if (currentSession === session && client) await refresh(); + }, report); + automaticRefresh = updates; try { - connected = await ImageClient.connect(address, (error) => { - if (currentSession !== session || disposed) return; - client = undefined; - tree.replace([]); - documents.disconnect(currentSession); - connectionStatus(false); - output.appendLine(`Disconnected from ${address}`); - if (error) report(error); - }); + connected = await ImageClient.connect( + address, + (error) => { + if (currentSession !== session || disposed) return; + client = undefined; + updates.dispose(); + tree.replace([]); + documents.disconnect(currentSession); + connectionStatus(false); + output.appendLine(`Disconnected from ${address}`); + if (error) report(error); + }, + () => updates.invalidate(), + ); } catch (error) { + updates.dispose(); if (currentSession === session) connectionStatus(false); throw error; } @@ -276,6 +290,7 @@ export function activate(context: vscode.ExtensionContext): void { { dispose: () => { disposed = true; + automaticRefresh?.dispose(); client?.close(); }, }, diff --git a/vscode/src/protocol.ts b/vscode/src/protocol.ts index f91e6fb..cf963e0 100644 --- a/vscode/src/protocol.ts +++ b/vscode/src/protocol.ts @@ -1,7 +1,8 @@ import { parse, stringify } from "lossless-json"; export const PROTOCOL_VERSION = 3; -export const MAX_LINE_BYTES = 1024 * 1024; +export const MAX_MESSAGE_BYTES = 1024 * 1024; +export const WEBSOCKET_SUBPROTOCOL = "runa-debug"; const ACTOR_REF = /^[A-Za-z0-9_-]{21}$/; export type Value = @@ -104,7 +105,7 @@ export function decodeValue(input: unknown): Value { case "string": { const data = text(value.data); bytes += Buffer.byteLength(data); - if (bytes > MAX_LINE_BYTES) throw new Error("Value exceeds string/key byte limit"); + if (bytes > MAX_MESSAGE_BYTES) throw new Error("Value exceeds string/key byte limit"); return { type, data }; } case "actor_ref": @@ -115,7 +116,7 @@ export function decodeValue(input: unknown): Value { case "record": { const entries = Object.entries(object(value.data)).map(([key, child]) => { bytes += Buffer.byteLength(key); - if (bytes > MAX_LINE_BYTES) throw new Error("Value exceeds string/key byte limit"); + if (bytes > MAX_MESSAGE_BYTES) throw new Error("Value exceeds string/key byte limit"); return [key, visit(child, depth + 1)]; }); return { type, data: Object.fromEntries(entries) }; @@ -163,14 +164,35 @@ function decodeActor(input: unknown): ActorSnapshot { export type ReadReply = { kind: "actors"; data: ActorSnapshot[] } | { kind: "actor"; data: ActorSnapshot }; -export function decodeResponse(line: string, requestId: string, command: ReadCommand): ReadReply { - const response = object( - parse(line, undefined, (literal) => { +function decodeEnvelope(message: string): Record { + const envelope = object( + parse(message, undefined, (literal) => { if (!/^-?(0|[1-9]\d*)$/.test(literal)) throw new Error("Non-integer protocol number"); return BigInt(literal); }), ); - if (response.version !== BigInt(PROTOCOL_VERSION)) throw new Error("Protocol version mismatch"); + if (envelope.version !== BigInt(PROTOCOL_VERSION)) throw new Error("Protocol version mismatch"); + return envelope; +} + +export function decodeNotification(message: string): bigint | undefined { + const envelope = decodeEnvelope(message); + if (!Object.hasOwn(envelope, "event")) return undefined; + if (Object.keys(envelope).length !== 2) throw new Error("Invalid notification envelope"); + const event = object(envelope.event); + if (event.event !== "image_changed" || Object.keys(event).length !== 2) { + throw new Error("Unsupported server event"); + } + const data = object(event.data); + const revision = uint(data.revision); + if (Object.keys(data).length !== 1 || revision === 0n) { + throw new Error("Invalid image revision"); + } + return revision; +} + +export function decodeResponse(line: string, requestId: string, command: ReadCommand): ReadReply { + const response = decodeEnvelope(line); if (response.request_id !== requestId) throw new Error("Response request ID mismatch"); const result = object(response.result); if (Object.keys(result).length !== 1) throw new Error("Invalid response result"); diff --git a/vscode/src/refresh.ts b/vscode/src/refresh.ts new file mode 100644 index 0000000..7362869 --- /dev/null +++ b/vscode/src/refresh.ts @@ -0,0 +1,40 @@ +/** Coalesce invalidations and rerun if a change arrives during an in-flight read. */ +export class RefreshQueue { + private dirty = false; + private running = false; + private stopped = false; + private timer?: ReturnType; + + constructor( + private readonly refresh: () => Promise, + private readonly report: (error: unknown) => void, + ) {} + + invalidate(): void { + if (this.stopped) return; + this.dirty = true; + if (this.running || this.timer) return; + this.timer = setTimeout(() => { + this.timer = undefined; + void this.run(); + }, 25); + } + + private async run(): Promise { + this.running = true; + this.dirty = false; + try { + await this.refresh(); + } catch (error) { + if (!this.stopped) this.report(error); + } finally { + this.running = false; + if (this.dirty) this.invalidate(); + } + } + + dispose(): void { + this.stopped = true; + clearTimeout(this.timer); + } +} diff --git a/vscode/test/client.test.ts b/vscode/test/client.test.ts index 1c6f3a2..e5a4a43 100644 --- a/vscode/test/client.test.ts +++ b/vscode/test/client.test.ts @@ -2,8 +2,8 @@ import assert from "node:assert/strict"; import { once } from "node:events"; import test from "node:test"; import { ImageClient, parseEndpoint } from "../src/client"; -import { MAX_LINE_BYTES, ProtocolError } from "../src/protocol"; -import { actor, fakeImage, json, REF, response } from "./fixtures"; +import { MAX_MESSAGE_BYTES, ProtocolError } from "../src/protocol"; +import { actor, fakeImage, json, notification, REF, response } from "./fixtures"; test("only numeric loopback endpoints and valid ports are accepted", () => { assert.deepEqual(parseEndpoint("127.0.0.1:3737"), { host: "127.0.0.1", port: 3737 }); @@ -36,8 +36,8 @@ test("fragmented UTF-8 replies and simultaneous reads preserve correlation and v : { kind: "actor", data: actor() }; const bytes = Buffer.from(response(request, reply)); const split = bytes.indexOf(Buffer.from("雪")) + 1; - socket.write(bytes.subarray(0, split)); - setImmediate(() => socket.write(bytes.subarray(split))); + socket.send(bytes.subarray(0, split), { binary: false, fin: false }); + setImmediate(() => socket.send(bytes.subarray(split), { binary: false, fin: true })); }); const client = await ImageClient.connect(address, (error) => assert.equal(error, undefined)); t.after(() => client.close()); @@ -51,13 +51,13 @@ test("protocol errors preserve the connection and are never retried", async (t) let requests = 0; const address = await fakeImage(t, (request, socket) => { requests++; - socket.write( + socket.send( requests === 1 ? json({ version: 3, request_id: request.request_id, result: { Err: { code: "not_found", message: "Missing actor" } }, - }) + "\n" + }) : response(request, { kind: "actors", data: [] }), ); }); @@ -69,23 +69,37 @@ test("protocol errors preserve the connection and are never retried", async (t) }); test("bad correlation, malformed JSON/UTF-8, oversize, truncation, and wrong versions close the stream", async (t) => { - for (const mode of ["correlation", "json", "utf8", "size", "truncated", "version", "kind"]) { + for (const mode of [ + "correlation", + "json", + "utf8", + "size", + "truncated", + "version", + "kind", + "binary", + "event", + ]) { await t.test(mode, async (t) => { let requests = 0; const failures: Error[] = []; const address = await fakeImage(t, (request, socket) => { requests++; - if (mode === "json") socket.write("{broken}\n"); - else if (mode === "utf8") socket.write(Buffer.from([0xff, 10])); - else if (mode === "size") socket.write(Buffer.alloc(MAX_LINE_BYTES + 1, 32)); - else if (mode === "truncated") socket.end('{"version":3'); + if (mode === "json") socket.send("{broken}"); + else if (mode === "utf8") socket.send(Buffer.from([0xff]), { binary: false }); + else if (mode === "size") socket.send(" ".repeat(MAX_MESSAGE_BYTES + 1)); + else if (mode === "truncated") { + socket.send('{"version":3', { fin: false }); + socket.terminate(); + } else if (mode === "binary") socket.send(Buffer.from("{}")); + else if (mode === "event") socket.send(json({ version: 3, event: { event: "unknown" } })); else - socket.write( + socket.send( json({ version: mode === "version" ? 2 : 3, request_id: mode === "correlation" ? "wrong" : request.request_id, result: { Ok: { kind: mode === "kind" ? "acknowledged" : "actors", data: [] } }, - }) + "\n", + }), ); }); const client = await ImageClient.connect(address, (error) => { @@ -105,7 +119,12 @@ test("request timeout rejects pending and queued reads without retry", async (t) const address = await fakeImage(t, () => { requests++; }); - const client = await ImageClient.connect(address, () => {}, { connect: 1000, request: 30 }); + const client = await ImageClient.connect( + address, + () => {}, + () => {}, + { connect: 1000, request: 30 }, + ); t.after(() => client.close()); await Promise.all([ assert.rejects(client.listActors(), /timed out/), @@ -116,8 +135,8 @@ test("request timeout rejects pending and queued reads without retry", async (t) test("the 1 MiB response limit is inclusive and framing survives the next request", async (t) => { const address = await fakeImage(t, (request, socket) => { - const line = response(request, { kind: "actors", data: [] }).slice(0, -1); - socket.write(line + " ".repeat(MAX_LINE_BYTES - Buffer.byteLength(line)) + "\n"); + const line = response(request, { kind: "actors", data: [] }); + socket.send(line + " ".repeat(MAX_MESSAGE_BYTES - Buffer.byteLength(line))); }); const client = await ImageClient.connect(address, (error) => assert.equal(error, undefined)); t.after(() => client.close()); @@ -125,10 +144,11 @@ test("the 1 MiB response limit is inclusive and framing survives the next reques assert.deepEqual(await client.listActors(), []); }); -test("unsolicited partial frames invalidate the connection", async (t) => { +test("unsolicited responses invalidate the connection", async (t) => { const failures: Error[] = []; const address = await fakeImage(t, (request, socket) => { - socket.write(response(request, { kind: "actors", data: [] }) + "{"); + socket.send(response(request, { kind: "actors", data: [] })); + socket.send(response(request, { kind: "actors", data: [] })); }); const client = await ImageClient.connect(address, (error) => { if (error) failures.push(error); @@ -140,7 +160,7 @@ test("unsolicited partial frames invalidate the connection", async (t) => { }); test("disconnect rejects pending reads and idle peer close is reported", async (t) => { - const address = await fakeImage(t, (_request, socket) => socket.end()); + const address = await fakeImage(t, (_request, socket) => socket.close()); let disconnected = false; const client = await ImageClient.connect(address, (error) => { assert.match(error?.message ?? "", /closed/); @@ -151,8 +171,8 @@ test("disconnect rejects pending reads and idle peer close is reported", async ( assert.equal(disconnected, true); const idleAddress = await fakeImage(t, (request, socket) => { - socket.write(response(request, { kind: "actors", data: [] })); - setImmediate(() => socket.end()); + socket.send(response(request, { kind: "actors", data: [] })); + setImmediate(() => socket.close()); }); let onClosed = () => {}; const closed = new Promise((resolve) => { @@ -165,6 +185,52 @@ test("disconnect rejects pending reads and idle peer close is reported", async ( await assert.rejects(idle.listActors(), /Not connected/); }); +test("requires the runa-debug WebSocket subprotocol", async (t) => { + const address = await fakeImage(t, () => {}, "wrong-protocol"); + await assert.rejects( + ImageClient.connect(address, () => {}), + /subprotocol/, + ); +}); + +test("notifications interleave with replies and arrive when no request is pending", async (t) => { + const revisions: bigint[] = []; + let onIdle = () => {}; + const idle = new Promise((resolve) => { + onIdle = resolve; + }); + const address = await fakeImage(t, (request, socket) => { + socket.send(notification(1n)); + socket.send(response(request, { kind: "actors", data: [] })); + setImmediate(() => socket.send(notification(5n))); + }); + const client = await ImageClient.connect( + address, + (error) => assert.equal(error, undefined), + (revision) => { + revisions.push(revision); + if (revision === 5n) onIdle(); + }, + ); + t.after(() => client.close()); + assert.deepEqual(await client.listActors(), []); + await idle; + assert.deepEqual(revisions, [1n, 5n]); +}); + +test("repeated or backwards revisions close the connection", async (t) => { + const failures: Error[] = []; + const address = await fakeImage(t, (_request, socket) => { + socket.send(notification(5n)); + socket.send(notification(5n)); + }); + const client = await ImageClient.connect(address, (error) => { + if (error) failures.push(error); + }); + t.after(() => client.close()); + await assert.rejects(client.listActors(), /Non-increasing/); + assert.equal(failures.length, 1); +}); test("failed connection reports the error", async () => { const { createServer } = await import("node:net"); const server = createServer(); diff --git a/vscode/test/fixtures.ts b/vscode/test/fixtures.ts index 46c70f8..44d22fa 100644 --- a/vscode/test/fixtures.ts +++ b/vscode/test/fixtures.ts @@ -1,8 +1,8 @@ -import { createServer, Socket } from "node:net"; -import { createInterface } from "node:readline"; +import { once } from "node:events"; +import WebSocket, { WebSocketServer } from "ws"; import { TestContext } from "node:test"; import { stringify } from "lossless-json"; -import { ActorSnapshot } from "../src/protocol"; +import { ActorSnapshot, WEBSOCKET_SUBPROTOCOL } from "../src/protocol"; export const REF = "abcdefghijklmnopqrstu"; @@ -42,26 +42,28 @@ export interface TestRequest { } export function response(request: TestRequest, reply: unknown): string { - return json({ version: 3, request_id: request.request_id, result: { Ok: reply } }) + "\n"; + return json({ version: 3, request_id: request.request_id, result: { Ok: reply } }); } export async function fakeImage( t: TestContext, - handle: (request: TestRequest, socket: Socket) => void, + handle: (request: TestRequest, socket: WebSocket) => void, + protocol = WEBSOCKET_SUBPROTOCOL, ): Promise { - const sockets = new Set(); - const server = createServer((socket) => { - sockets.add(socket); - socket.on("close", () => sockets.delete(socket)); + const server = new WebSocketServer({ + host: "127.0.0.1", + port: 0, + handleProtocols: (offered) => (offered.has(protocol) ? protocol : false), + }); + server.on("connection", (socket) => { socket.on("error", () => { // Transport-failure tests intentionally close a socket while the fixture is writing. }); - const lines = createInterface({ input: socket }); - lines.on("line", (line) => handle(JSON.parse(line), socket)); + socket.on("message", (data) => handle(JSON.parse(data.toString()), socket)); }); - await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); + await once(server, "listening"); t.after(async () => { - for (const socket of sockets) socket.destroy(); + for (const socket of server.clients) socket.terminate(); await new Promise((resolve, reject) => server.close((error) => (error ? reject(error) : resolve())), ); @@ -70,3 +72,7 @@ export async function fakeImage( if (!address || typeof address === "string") throw new Error("Missing test listener"); return `127.0.0.1:${address.port}`; } + +export function notification(revision: bigint): string { + return json({ version: 3, event: { event: "image_changed", data: { revision } } }); +} diff --git a/vscode/test/protocol.test.ts b/vscode/test/protocol.test.ts index 97e7d4c..0795949 100644 --- a/vscode/test/protocol.test.ts +++ b/vscode/test/protocol.test.ts @@ -3,6 +3,7 @@ import test from "node:test"; import { parse } from "lossless-json"; import { decodeResponse, + decodeNotification, decodeValue, formatSnapshot, ProtocolError, @@ -14,6 +15,41 @@ const command: ReadCommand = { operation: "list_actors" }; const envelope = (reply: unknown) => json({ version: 3, request_id: "test", result: { Ok: reply } }); +test("notification envelopes preserve u64 revisions and reject malformed events", () => { + const notification = (data: unknown) => + json({ + version: 3, + event: { + event: "image_changed", + data, + }, + }); + assert.equal( + decodeNotification(notification({ revision: 18446744073709551615n })), + 18446744073709551615n, + ); + assert.equal(decodeNotification(envelope({ kind: "actors", data: [] })), undefined); + for (const data of [ + { revision: 0 }, + { revision: -1 }, + { revision: 1.5 }, + { revision: 18446744073709551616n }, + { revision: "1" }, + { revision: 1, extra: true }, + {}, + ]) { + assert.throws(() => decodeNotification(notification(data))); + } + assert.throws(() => + decodeNotification( + json({ + version: 2, + event: { event: "image_changed", data: { revision: 1 } }, + }), + ), + ); +}); + test("v3 snapshots preserve all value variants, i64 limits, u64 telemetry, and mailbox order", () => { const snapshot = actor(); const reply = decodeResponse(envelope({ kind: "actors", data: [snapshot] }), "test", command); diff --git a/vscode/test/refresh.test.ts b/vscode/test/refresh.test.ts new file mode 100644 index 0000000..060d876 --- /dev/null +++ b/vscode/test/refresh.test.ts @@ -0,0 +1,53 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import { RefreshQueue } from "../src/refresh"; + +test("refreshes coalesce bursts and rerun after in-flight invalidation", async (t) => { + const reads: (() => void)[] = []; + const queue = new RefreshQueue( + () => + new Promise((resolve) => { + reads.push(resolve); + }), + (error) => { + throw error; + }, + ); + t.after(() => queue.dispose()); + for (let i = 0; i < 100; i++) queue.invalidate(); + await waitFor(() => reads.length === 1); + for (let i = 0; i < 100; i++) queue.invalidate(); + assert.equal(reads.length, 1); + reads[0]?.(); + await waitFor(() => reads.length === 2); + reads[1]?.(); + await new Promise((resolve) => setTimeout(resolve, 60)); + assert.equal(reads.length, 2); +}); + +test("disposed sessions never run queued refreshes and errors are reported", async () => { + let calls = 0; + const errors: unknown[] = []; + const queue = new RefreshQueue( + async () => { + calls++; + throw new Error("refresh failed"); + }, + (error) => errors.push(error), + ); + queue.invalidate(); + await waitFor(() => errors.length === 1); + assert.equal(calls, 1); + queue.invalidate(); + queue.dispose(); + await new Promise((resolve) => setTimeout(resolve, 60)); + assert.equal(calls, 1); +}); + +async function waitFor(condition: () => boolean): Promise { + const deadline = Date.now() + 2000; + while (!condition()) { + if (Date.now() > deadline) throw new Error("Refresh queue did not make progress"); + await new Promise((resolve) => setTimeout(resolve, 5)); + } +}