From 4215939207c9dff5b613d997e9baf2beb0c0929c Mon Sep 17 00:00:00 2001 From: Mirko Brombin Date: Tue, 1 Sep 2026 15:29:04 +0000 Subject: [PATCH 1/3] feat: add transparent Warp StreamBus transport --- CHANGELOG.md | 9 + README.md | 4 +- cmd/rfw/build/build.go | 14 +- cmd/rfw/build/build_test.go | 9 +- cmd/rfw/build/delivery.go | 36 +- cmd/rfw/build/delivery_test.go | 18 +- cmd/rfw/initproj/template/rfw.json | 1 + docs/articles/guide/ssc.md | 3 +- docs/articles/guide/transports.md | 88 +++++ examples/counter/rfw.json | 1 + examples/dashboard/rfw.json | 1 + examples/dynamic-list/rfw.json | 1 + go.mod | 11 +- go.sum | 32 +- host/framing.go | 46 +++ host/framing_test.go | 22 ++ host/server.go | 22 +- host/session.go | 3 + host/streambus_benchmark_test.go | 103 ++++++ host/streambus_integration_test.go | 107 ++++++ host/streambus_server.go | 389 ++++++++++++++++++++ host/transport.go | 53 +++ host/transport_test.go | 48 +++ host/websocket.go | 22 +- hostclient/streambus_transport_wasm.go | 268 ++++++++++++++ hostclient/streambus_transport_wasm_test.go | 38 ++ hostclient/transport_selector.go | 8 + 27 files changed, 1329 insertions(+), 28 deletions(-) create mode 100644 docs/articles/guide/transports.md create mode 100644 host/framing.go create mode 100644 host/framing_test.go create mode 100644 host/streambus_benchmark_test.go create mode 100644 host/streambus_integration_test.go create mode 100644 host/streambus_server.go create mode 100644 host/transport.go create mode 100644 host/transport_test.go create mode 100644 hostclient/streambus_transport_wasm.go create mode 100644 hostclient/streambus_transport_wasm_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index eb42870c..5913b05a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,14 @@ # Changelog +## Unreleased + +- Add transparent Warp StreamBus transport selection through the top-level + `transport` manifest key or `RFW_TRANSPORT`. +- Serve bounded StreamBus queues over WebTransport and HTTP/3 while preserving + WebSocket as an automatic compatibility fallback. +- Keep SSC components, actions, hydration, sessions, replay and broadcasts + transport-independent. + All notable changes to this project are documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), diff --git a/README.md b/README.md index 78711e81..918a4329 100644 --- a/README.md +++ b/README.md @@ -17,7 +17,7 @@ rfw is "Phoenix LiveView for Go". It lets you build interactive, real-time web apps using Server Side Computed (SSC) components. -Instead of writing a REST API and a frontend framework, you write Go. rfw handles the WebSocket synchronization and DOM updates for you. It is ideal for: +Instead of writing a REST API and a frontend framework, you write Go. rfw handles synchronization over WebSocket or Warp StreamBus and updates the DOM for you. It is ideal for: - Real-time dashboards - Internal admin tools - Control planes @@ -125,7 +125,7 @@ Using rfw in production or in a side project? Open a PR and add yourself here. ## Server Side Computed (SSC) -SSC is the core of rfw. Most application logic runs on the server, while the browser loads a lightweight binary to hydrate the HTML. The server and client keep state synchronized through a persistent WebSocket connection. +SSC is the core of rfw. Most application logic runs on the server, while the browser loads a lightweight binary to hydrate the HTML. The server and client keep state synchronized through WebSocket or Warp StreamBus over WebTransport. The transport is selected through `rfw.json` or `RFW_TRANSPORT` without changing application code. Components use host signal types (`t.HInt`, `t.HString`, etc.) to declare server-synced bindings. See the [SSC guide](./docs/articles/guide/ssc.md) for more details. diff --git a/cmd/rfw/build/build.go b/cmd/rfw/build/build.go index d998c895..804e5179 100644 --- a/cmd/rfw/build/build.go +++ b/cmd/rfw/build/build.go @@ -48,7 +48,7 @@ func pluginsConfig(explicit map[string]json.RawMessage) map[string]json.RawMessa // Build compiles the configured application and runs its build plugins. func Build() error { - shape := buildShape{delivery: deliveryNetwork, transport: sscTransportBrowser} + shape := buildShape{delivery: deliveryNetwork, transport: sscTransportBrowser, hostWire: hostTransportWebSocket} if data, readErr := os.ReadFile("rfw.json"); readErr == nil { decoded, err := decodeBuildShape(data) if err != nil { @@ -56,6 +56,13 @@ func Build() error { } shape = decoded } + if configured := strings.TrimSpace(os.Getenv("RFW_TRANSPORT")); configured != "" { + transport, err := parseHostTransport(configured) + if err != nil { + return err + } + shape.hostWire = transport + } if err := plugins.Configure(pluginsConfig(shape.plugins)); err != nil { return fmt.Errorf("failed to configure plugins: %w", err) } @@ -166,7 +173,7 @@ func Build() error { // is served by something that will not, so the client picks an artifact by // name instead, and an embedded build is read from the container's local // asset handler, which negotiates nothing either. - if err := writeClientConfig(clientDir, shape.host, version, encodings, shape.negotiates(), !isDev, shape.delivery, shape.transport); err != nil { + if err := writeClientConfig(clientDir, shape.host, version, encodings, shape.negotiates(), !isDev, shape.delivery, shape.transport, shape.hostWire); err != nil { return fmt.Errorf("failed to write client config: %w", err) } if err := stampIndexHTML(filepath.Join(clientDir, "index.html"), version); err != nil { @@ -304,7 +311,7 @@ func readWasmExec() ([]byte, error) { // bundle it has a compressed copy of. That is also why the delivery mode is // stamped explicitly instead of being inferred from an empty encoding list, // which a failed compression step would produce too. -func writeClientConfig(clientDir, host, wasmVersion string, encodings []Encoding, negotiated, production bool, deliveryMode delivery, transport sscTransport) error { +func writeClientConfig(clientDir, host, wasmVersion string, encodings []Encoding, negotiated, production bool, deliveryMode delivery, transport sscTransport, hostWire hostTransport) error { var b strings.Builder b.WriteString("// Generated by rfw build. Do not edit.\n") if h := strings.TrimSpace(host); h != "" { @@ -319,6 +326,7 @@ func writeClientConfig(clientDir, host, wasmVersion string, encodings []Encoding fmt.Fprintf(&b, "window.RFW_WASM_ENCODINGS = [%s];\n", strings.Join(names, ", ")) fmt.Fprintf(&b, "window.RFW_WASM_NEGOTIATED = %t;\n", negotiated) fmt.Fprintf(&b, "window.RFW_SSC_TRANSPORT = %q;\n", string(transport)) + fmt.Fprintf(&b, "window.RFW_TRANSPORT = %q;\n", string(hostWire)) mode := "development" if production { mode = "production" diff --git a/cmd/rfw/build/build_test.go b/cmd/rfw/build/build_test.go index 82ce678b..b24b777a 100644 --- a/cmd/rfw/build/build_test.go +++ b/cmd/rfw/build/build_test.go @@ -329,7 +329,7 @@ func TestWriteClientConfig(t *testing.T) { dir := t.TempDir() t.Chdir(dir) encodings := []Encoding{EncodingBrotli, EncodingGzip} - if err := writeClientConfig(".", "wss://example.com/rfw", "abc123", encodings, true, true, deliveryNetwork, sscTransportBrowser); err != nil { + if err := writeClientConfig(".", "wss://example.com/rfw", "abc123", encodings, true, true, deliveryNetwork, sscTransportBrowser, hostTransportStreamBus); err != nil { t.Fatalf("writeClientConfig: %v", err) } @@ -345,6 +345,7 @@ func TestWriteClientConfig(t *testing.T) { `window.RFW_WASM_ENCODINGS = ["br", "gzip"];`, `window.RFW_WASM_NEGOTIATED = true;`, `window.RFW_SSC_TRANSPORT = "browser";`, + `window.RFW_TRANSPORT = "streambus";`, `window.RFW_BUILD_MODE = "production";`, } { if !strings.Contains(got, want) { @@ -359,7 +360,7 @@ func TestWriteClientConfig(t *testing.T) { func TestWriteClientConfigDescribesAnUncompressedBuild(t *testing.T) { dir := t.TempDir() t.Chdir(dir) - if err := writeClientConfig(".", "", "devbuild", nil, false, false, deliveryNetwork, sscTransportBrowser); err != nil { + if err := writeClientConfig(".", "", "devbuild", nil, false, false, deliveryNetwork, sscTransportBrowser, hostTransportWebSocket); err != nil { t.Fatalf("writeClientConfig: %v", err) } config, err := os.ReadFile("rfw_config.js") @@ -387,7 +388,7 @@ func TestWriteClientConfigDescribesAnUncompressedBuild(t *testing.T) { func TestWriteClientConfigDescribesAnEmbeddedBuild(t *testing.T) { dir := t.TempDir() t.Chdir(dir) - if err := writeClientConfig(".", "wss://api.example.com/ws", "abc123", nil, false, true, deliveryEmbedded, sscTransportCapacitor); err != nil { + if err := writeClientConfig(".", "wss://api.example.com/ws", "abc123", nil, false, true, deliveryEmbedded, sscTransportCapacitor, hostTransportWebSocket); err != nil { t.Fatalf("writeClientConfig: %v", err) } config, err := os.ReadFile("rfw_config.js") @@ -415,7 +416,7 @@ func TestWriteClientConfigDescribesAnEmbeddedBuild(t *testing.T) { func TestWriteClientConfigDefinesEveryGlobalTheLoaderReads(t *testing.T) { dir := t.TempDir() t.Chdir(dir) - if err := writeClientConfig(".", "", "abc123", []Encoding{EncodingGzip}, false, true, deliveryNetwork, sscTransportBrowser); err != nil { + if err := writeClientConfig(".", "", "abc123", []Encoding{EncodingGzip}, false, true, deliveryNetwork, sscTransportBrowser, hostTransportWebSocket); err != nil { t.Fatalf("writeClientConfig: %v", err) } config, err := os.ReadFile("rfw_config.js") diff --git a/cmd/rfw/build/delivery.go b/cmd/rfw/build/delivery.go index c29e28c6..973eedc4 100644 --- a/cmd/rfw/build/delivery.go +++ b/cmd/rfw/build/delivery.go @@ -23,6 +23,11 @@ type delivery string // origin boundary. type sscTransport string +// hostTransport names the browser-to-host wire transport used by SSC. It is +// orthogonal to sscTransport: the browser runtime can use WebSocket or +// StreamBus, while the Capacitor runtime keeps using its native bridge. +type hostTransport string + const ( // deliveryNetwork is the default. The bundle is downloaded, so the build // writes compressed artifacts and the loader picks one. @@ -36,6 +41,10 @@ const ( sscTransportBrowser sscTransport = "browser" sscTransportCapacitor sscTransport = "capacitor" + + hostTransportWebSocket hostTransport = "websocket" + hostTransportStreamBus hostTransport = "streambus" + hostTransportAuto hostTransport = "auto" ) // parseDelivery resolves the build.delivery value from rfw.json. An empty value @@ -72,6 +81,22 @@ func parseSSCTransport(value string) (sscTransport, error) { } } +func parseHostTransport(value string) (hostTransport, error) { + switch transport := hostTransport(strings.ToLower(strings.TrimSpace(value))); transport { + case "", hostTransportWebSocket, "ws": + return hostTransportWebSocket, nil + case hostTransportStreamBus, "webtransport", "warp-streambus": + return hostTransportStreamBus, nil + case hostTransportAuto: + return hostTransportAuto, nil + default: + return "", fmt.Errorf( + "rfw.json: transport %q is not a host transport; use %q, %q or %q", + value, hostTransportWebSocket, hostTransportStreamBus, hostTransportAuto, + ) + } +} + // buildShape is the delivery-relevant view of rfw.json. It exists so the two // settings that shape a build stay separable: "static" decides whether the SSC // client is linked, "embedded" decides how the bundle is delivered, and neither @@ -81,6 +106,7 @@ type buildShape struct { host string delivery delivery transport sscTransport + hostWire hostTransport plugins map[string]json.RawMessage } @@ -102,7 +128,8 @@ func (s buildShape) compresses() bool { // unknown delivery mode fails the build rather than quietly ignoring it. func decodeBuildShape(data []byte) (buildShape, error) { var manifest struct { - Build struct { + Transport string `json:"transport"` + Build struct { Type string `json:"type"` Host string `json:"host"` Delivery string `json:"delivery"` @@ -111,7 +138,7 @@ func decodeBuildShape(data []byte) (buildShape, error) { Plugins map[string]json.RawMessage `json:"plugins"` } if err := json.Unmarshal(data, &manifest); err != nil { - return buildShape{delivery: deliveryNetwork, transport: sscTransportBrowser}, nil + return buildShape{delivery: deliveryNetwork, transport: sscTransportBrowser, hostWire: hostTransportWebSocket}, nil } mode, err := parseDelivery(manifest.Build.Delivery) if err != nil { @@ -121,11 +148,16 @@ func decodeBuildShape(data []byte) (buildShape, error) { if err != nil { return buildShape{}, err } + hostWire, err := parseHostTransport(manifest.Transport) + if err != nil { + return buildShape{}, err + } return buildShape{ static: manifest.Build.Type == "static", host: manifest.Build.Host, delivery: mode, transport: transport, + hostWire: hostWire, plugins: manifest.Plugins, }, nil } diff --git a/cmd/rfw/build/delivery_test.go b/cmd/rfw/build/delivery_test.go index 62f223c1..8a0e8a0a 100644 --- a/cmd/rfw/build/delivery_test.go +++ b/cmd/rfw/build/delivery_test.go @@ -114,7 +114,12 @@ func TestDecodeBuildShapeKeepsDeliveryAndTypeSeparate(t *testing.T) { { name: "embedded SSC can select the Capacitor transport", manifest: `{"build":{"type":"ssc","host":"wss://api.example.com/ws","delivery":"embedded","sscTransport":"capacitor"}}`, - want: buildShape{host: "wss://api.example.com/ws", delivery: deliveryEmbedded, transport: sscTransportCapacitor}, + want: buildShape{host: "wss://api.example.com/ws", delivery: deliveryEmbedded, transport: sscTransportCapacitor, hostWire: hostTransportWebSocket}, + }, + { + name: "SSC can select StreamBus independently from its runtime", + manifest: `{"transport":"streambus","build":{"type":"ssc","host":"https://api.example.com"}}`, + want: buildShape{host: "https://api.example.com", delivery: deliveryNetwork, transport: sscTransportBrowser, hostWire: hostTransportStreamBus}, }, } { t.Run(tc.name, func(t *testing.T) { @@ -134,6 +139,9 @@ func TestDecodeBuildShapeKeepsDeliveryAndTypeSeparate(t *testing.T) { if got.transport != tc.want.transport { t.Fatalf("transport = %q, want %q", got.transport, tc.want.transport) } + if got.hostWire != tc.want.hostWire && tc.want.hostWire != "" { + t.Fatalf("host wire transport = %q, want %q", got.hostWire, tc.want.hostWire) + } }) } } @@ -150,6 +158,12 @@ func TestDecodeBuildShapeRejectsAnUnknownSSCTransport(t *testing.T) { } } +func TestDecodeBuildShapeRejectsAnUnknownHostTransport(t *testing.T) { + if _, err := decodeBuildShape([]byte(`{"transport":"telepathy"}`)); err == nil { + t.Fatal("an unknown host transport built without an error") + } +} + // An unreadable manifest has always fallen back to the defaults rather than // failing the build, and it still does. func TestDecodeBuildShapeToleratesAnUnparsableManifest(t *testing.T) { @@ -157,7 +171,7 @@ func TestDecodeBuildShapeToleratesAnUnparsableManifest(t *testing.T) { if err != nil { t.Fatalf("decodeBuildShape: %v", err) } - if got.delivery != deliveryNetwork || got.transport != sscTransportBrowser || got.static || got.plugins != nil { + if got.delivery != deliveryNetwork || got.transport != sscTransportBrowser || got.hostWire != hostTransportWebSocket || got.static || got.plugins != nil { t.Fatalf("shape = %+v, want the defaults", got) } } diff --git a/cmd/rfw/initproj/template/rfw.json b/cmd/rfw/initproj/template/rfw.json index 816b2ffd..34862bb9 100644 --- a/cmd/rfw/initproj/template/rfw.json +++ b/cmd/rfw/initproj/template/rfw.json @@ -1,4 +1,5 @@ { + "transport": "websocket", "build": { "type": "ssc", "host": "" diff --git a/docs/articles/guide/ssc.md b/docs/articles/guide/ssc.md index 87d370a8..d4c1d1df 100644 --- a/docs/articles/guide/ssc.md +++ b/docs/articles/guide/ssc.md @@ -2,7 +2,8 @@ SSC splits a component in two: an HTML component that renders in the browser, and a host component that runs in your Go server process. A -persistent WebSocket at `/ws` keeps them synchronized. The browser loads a +persistent transport keeps them synchronized. WebSocket remains available at +`/ws`, while Warp StreamBus can run over WebTransport at `/streambus`. The browser loads a lightweight wasm binary that hydrates the HTML; your logic, queries and secrets stay on the server. For the trust boundary and hardening options, read the [SSC security model](ssc-security.md). diff --git a/docs/articles/guide/transports.md b/docs/articles/guide/transports.md new file mode 100644 index 00000000..cfdd0988 --- /dev/null +++ b/docs/articles/guide/transports.md @@ -0,0 +1,88 @@ +# SSC transports + +RFW keeps the SSC API independent from its network transport. Components, +`hostclient.Send`, typed actions, hydration, session resumption and broadcasts +behave the same with WebSocket and Warp StreamBus. + +## Selecting StreamBus + +Set the top-level `transport` key in `rfw.json`: + +```json +{ + "transport": "streambus", + "build": { + "type": "ssc", + "host": "" + } +} +``` + +The environment overrides the manifest: + +```bash +RFW_TRANSPORT=streambus rfw dev +``` + +Accepted values are: + +| Value | Behaviour | +| --- | --- | +| `websocket` | Use the existing `/ws` endpoint. This remains the default. | +| `streambus` | Prefer Warp StreamBus over WebTransport and fall back to WebSocket if the browser or network cannot establish it. | +| `auto` | Probe StreamBus first and transparently use WebSocket when unavailable. | + +`webtransport` and `warp-streambus` are accepted aliases for `streambus`. + +## Runtime model + +StreamBus runs on WebTransport over HTTP/3 at `/streambus`. RFW preserves its +existing JSON SSC protocol and length-prefixes messages on a reliable QUIC +stream. On the server, Warp StreamBus provides bounded queues, priorities, +replay storage and explicit backpressure before frames reach the network. + +The WebSocket endpoint remains mounted as a compatibility fallback. No +application code or component API changes when the selected transport changes. + +Production deployments must route TCP and UDP for the HTTPS port and use a +certificate trusted by the browser. During `rfw dev`, the HTTP/3 endpoint uses +the HTTPS development port, which is the configured HTTP port plus one. RFW +automatically exposes the short-lived development certificate hash to the WASM +client so WebTransport can validate the self-signed certificate without an +application-level exception. + +Set `window.RFW_STREAMBUS_URL` before loading the WASM module only when the +HTTP/3 endpoint uses a different public URL. Otherwise RFW derives it from +`build.host` or the page origin. + +## Performance scope + +The checked-in fan-out benchmark measures producer-side dispatch cost. It +compares StreamBus publication into bounded subscriber queues with RFW's +synchronous WebSocket fallback send path. Production WebSocket endpoints also +have an outbound queue, so this is deliberately not an end-to-end latency or +network-throughput claim. + +Run it with: + +```bash +go test ./host -run '^$' -bench 'Benchmark(StreamBus|WebSocket)Fanout' -benchmem -benchtime=2s -count=3 +``` + +These numbers do not claim that every individual WebTransport round trip is +faster than every WebSocket round trip. They demonstrate the improvement for +the heavy fan-out and burst workloads StreamBus is designed to handle. + +Reference run on Go 1.25.0 and an AMD EPYC 9V74, median of three 500 ms runs: + +| Subscribers | StreamBus | WebSocket | Producer speedup | +| ---: | ---: | ---: | ---: | +| 1 | 878 ns/op | 4,483 ns/op | 5.1x | +| 10 | 10,505 ns/op | 47,526 ns/op | 4.5x | +| 100 | 115,784 ns/op | 507,769 ns/op | 4.4x | + +At 100 subscribers, StreamBus used 28,801 B and 500 allocations per broadcast, +compared with 112,964 B and 1,801 allocations for synchronous WebSocket +fan-out. That is about 3.9x less allocated memory and 3.6x fewer allocations on +the producer path. The StreamBus benchmark includes the distinct SSC JSON +serialization and queue publish required for every target session. diff --git a/examples/counter/rfw.json b/examples/counter/rfw.json index b7ab8138..d0a348cb 100644 --- a/examples/counter/rfw.json +++ b/examples/counter/rfw.json @@ -1,4 +1,5 @@ { + "transport": "websocket", "build": { "type": "static" } diff --git a/examples/dashboard/rfw.json b/examples/dashboard/rfw.json index b7ab8138..d0a348cb 100644 --- a/examples/dashboard/rfw.json +++ b/examples/dashboard/rfw.json @@ -1,4 +1,5 @@ { + "transport": "websocket", "build": { "type": "static" } diff --git a/examples/dynamic-list/rfw.json b/examples/dynamic-list/rfw.json index b7ab8138..d0a348cb 100644 --- a/examples/dynamic-list/rfw.json +++ b/examples/dynamic-list/rfw.json @@ -1,4 +1,5 @@ { + "transport": "websocket", "build": { "type": "static" } diff --git a/go.mod b/go.mod index 4158ad68..40a6385b 100644 --- a/go.mod +++ b/go.mod @@ -8,19 +8,26 @@ require ( github.com/fatih/color v1.19.0 github.com/fsnotify/fsnotify v1.8.0 github.com/mirkobrombin/go-cli-builder v1.0.0 - github.com/mirkobrombin/go-foundation/v2 v2.1.0 + github.com/mirkobrombin/go-foundation/v2 v2.4.0 github.com/mirkobrombin/go-logger v0.2.0 github.com/mirkobrombin/go-signal/v2 v2.0.0 + github.com/mirkobrombin/go-warp/v2 v2.1.0 github.com/pkg/browser v0.0.0-20240102092130-5ac0b6a4141c + github.com/quic-go/quic-go v0.59.0 + github.com/quic-go/webtransport-go v0.10.0 github.com/russross/blackfriday/v2 v2.1.0 github.com/tdewolff/minify/v2 v2.24.3 - golang.org/x/net v0.57.0 + golang.org/x/net v0.58.0 ) require ( + github.com/dunglas/httpsfv v1.1.0 // indirect github.com/mattn/go-colorable v0.1.14 // indirect github.com/mattn/go-isatty v0.0.20 // indirect + github.com/quic-go/qpack v0.6.0 // indirect github.com/tdewolff/parse/v2 v2.8.3 // indirect + golang.org/x/crypto v0.55.0 // indirect golang.org/x/sys v0.47.0 // indirect + golang.org/x/text v0.41.0 // indirect gopkg.in/natefinch/lumberjack.v2 v2.2.1 // indirect ) diff --git a/go.sum b/go.sum index a6f742a2..61699c3f 100644 --- a/go.sum +++ b/go.sum @@ -2,6 +2,10 @@ github.com/andybalholm/brotli v1.2.1 h1:R+f5xP285VArJDRgowrfb9DqL18yVK0gKAW/F+eT github.com/andybalholm/brotli v1.2.1/go.mod h1:rzTDkvFWvIrjDXZHkuS16NPggd91W3kUSvPlQ1pLaKY= github.com/bmatcuk/doublestar/v4 v4.10.0 h1:zU9WiOla1YA122oLM6i4EXvGW62DvKZVxIe6TYWexEs= github.com/bmatcuk/doublestar/v4 v4.10.0/go.mod h1:xBQ8jztBU6kakFMg+8WGxn0c6z1fTSPVIjEY1Wr7jzc= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/dunglas/httpsfv v1.1.0 h1:Jw76nAyKWKZKFrpMMcL76y35tOpYHqQPzHQiwDvpe54= +github.com/dunglas/httpsfv v1.1.0/go.mod h1:zID2mqw9mFsnt7YC3vYQ9/cjq30q41W+1AnDwH8TiMg= github.com/fatih/color v1.19.0 h1:Zp3PiM21/9Ld6FzSKyL5c/BULoe/ONr9KlbYVOfG8+w= github.com/fatih/color v1.19.0/go.mod h1:zNk67I0ZUT1bEGsSGyCZYZNrHuTkJJB+r6Q9VuMi0LE= github.com/fsnotify/fsnotify v1.8.0 h1:dAwr6QBTBZIkG8roQaJjGof0pp0EeF+tNV7YBP3F/8M= @@ -12,16 +16,28 @@ github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWE github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y= github.com/mirkobrombin/go-cli-builder v1.0.0 h1:oLFch1y2HIaZ6MqwKGwDayXXhQfC3HAjrLpC/7xCrr8= github.com/mirkobrombin/go-cli-builder v1.0.0/go.mod h1:fOd36mpPgiOZubwnBaoKU4MMwZpdlsWZYulyHsYgNY8= -github.com/mirkobrombin/go-foundation/v2 v2.1.0 h1:AliJbX23Uhs29PlybYd2/rJT7YE5tm2Pq7fLboD+28k= -github.com/mirkobrombin/go-foundation/v2 v2.1.0/go.mod h1:fYCg04t/V9/vJ10itHY9kk6qbix7D1POxBoHec0x1CE= +github.com/mirkobrombin/go-foundation/v2 v2.4.0 h1:KQgGj0A+rnWiLaIPx16P2aoyijn6pi7UNmrBXvMH5j8= +github.com/mirkobrombin/go-foundation/v2 v2.4.0/go.mod h1:j/zVrYN/o6aFtkXs+MBNjC/JSEG3YS1/TBKOcnc3aTs= github.com/mirkobrombin/go-logger v0.2.0 h1:qAdo9oCIAvgGOuQIRah5kbFQa3IoHxSjgwYdP3bA6Xw= github.com/mirkobrombin/go-logger v0.2.0/go.mod h1:6YeeNrLAEswz0W5eJZ9AWhO/KpFaunznHWzUhpG4jRE= github.com/mirkobrombin/go-signal/v2 v2.0.0 h1:8XHnop9mdysg+Mlk1KYKkbIhDk3I6TQHh7dJh8aJo9E= github.com/mirkobrombin/go-signal/v2 v2.0.0/go.mod h1:Csmo2dYGH8or5nVVq4TMydv/LGXuv+Wg0Ryj5XsNWlg= +github.com/mirkobrombin/go-warp/v2 v2.1.0 h1:RPRumUxq7bW5QkbJ7zCkIDYDCN1crcOQTyT5OKTUcvw= +github.com/mirkobrombin/go-warp/v2 v2.1.0/go.mod h1:52d4cz1LlCJPGHc6oOdDFBuouFd4RZ7wFLGr87qkh0I= github.com/pkg/browser v0.0.0-20240102092130-5ac0b6a4141c h1:+mdjkGKdHQG3305AYmdv1U2eRNDiU2ErMBj1gwrq8eQ= github.com/pkg/browser v0.0.0-20240102092130-5ac0b6a4141c/go.mod h1:7rwL4CYBLnjLxUqIJNnCWiEdr3bn6IUYi15bNlnbCCU= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/quic-go/qpack v0.6.0 h1:g7W+BMYynC1LbYLSqRt8PBg5Tgwxn214ZZR34VIOjz8= +github.com/quic-go/qpack v0.6.0/go.mod h1:lUpLKChi8njB4ty2bFLX2x4gzDqXwUpaO1DP9qMDZII= +github.com/quic-go/quic-go v0.59.0 h1:OLJkp1Mlm/aS7dpKgTc6cnpynnD2Xg7C1pwL6vy/SAw= +github.com/quic-go/quic-go v0.59.0/go.mod h1:upnsH4Ju1YkqpLXC305eW3yDZ4NfnNbmQRCMWS58IKU= +github.com/quic-go/webtransport-go v0.10.0 h1:LqXXPOXuETY5Xe8ITdGisBzTYmUOy5eSj+9n4hLTjHI= +github.com/quic-go/webtransport-go v0.10.0/go.mod h1:LeGIXr5BQKE3UsynwVBeQrU1TPrbh73MGoC6jd+V7ow= github.com/russross/blackfriday/v2 v2.1.0 h1:JIOH55/0cWyOuilr9/qlrm0BSXldqnqwMsf35Ld67mk= github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM= +github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= +github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= github.com/tdewolff/minify/v2 v2.24.3 h1:BaKgWSFLKbKDiUskbeRgbe2n5d1Ci1x3cN/eXna8zOA= github.com/tdewolff/minify/v2 v2.24.3/go.mod h1:1JrCtoZXaDbqioQZfk3Jdmr0GPJKiU7c1Apmb+7tCeE= github.com/tdewolff/parse/v2 v2.8.3 h1:5VbvtJ83cfb289A1HzRA9sf02iT8YyUwN84ezjkdY1I= @@ -30,11 +46,19 @@ github.com/tdewolff/test v1.0.11 h1:FdLbwQVHxqG16SlkGveC0JVyrJN62COWTRyUFzfbtBE= github.com/tdewolff/test v1.0.11/go.mod h1:XPuWBzvdUzhCuxWO1ojpXsyzsA5bFoS3tO/Q3kFuTG8= github.com/xyproto/randomstring v1.0.5 h1:YtlWPoRdgMu3NZtP45drfy1GKoojuR7hmRcnhZqKjWU= github.com/xyproto/randomstring v1.0.5/go.mod h1:rgmS5DeNXLivK7YprL0pY+lTuhNQW3iGxZ18UQApw/E= -golang.org/x/net v0.57.0 h1:K5+3DljvIuDG9/Jv9rvyMywYNFCQ9RSUY6OOTTkT+tE= -golang.org/x/net v0.57.0/go.mod h1:KpXc8iv+r3XplLAG/f7Jsf9RPszJzdR0f58q9vGOuEU= +go.uber.org/mock v0.5.2 h1:LbtPTcP8A5k9WPXj54PPPbjcI4Y6lhyOZXn+VS7wNko= +go.uber.org/mock v0.5.2/go.mod h1:wLlUxC2vVTPTaE3UD51E0BGOAElKrILxhVSDYQLld5o= +golang.org/x/crypto v0.55.0 h1:+KWHjbgOaAQ66dh/YlkZKHlz9ZUlq61AFirAR9ntP8M= +golang.org/x/crypto v0.55.0/go.mod h1:uq0V9dE/fzQuJtbnL+2EhWOE63vo164FY8xqEnV9xis= +golang.org/x/net v0.58.0 h1:ynWG7rqYi4ccpTEuPZ2QGWHktVEM9DMCj9yzDE0Q7To= +golang.org/x/net v0.58.0/go.mod h1:YwCddHnFlT7eLQqVprV19OnhLGtc5xOKgE0RyqgfWAU= golang.org/x/sys v0.1.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs= golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +golang.org/x/text v0.41.0 h1:vz/seA0lnX87Othu2f/0L24RcgrXD9/YFTSuGjj3rH8= +golang.org/x/text v0.41.0/go.mod h1:jvf1O8ajNzZqhSrQBPbutR/EB83Cc0CFrezNQIwbb5M= gopkg.in/natefinch/lumberjack.v2 v2.2.1 h1:bBRl1b0OH9s/DuPhuXpNl+VtCaJXFZ5/uEFST95x9zc= gopkg.in/natefinch/lumberjack.v2 v2.2.1/go.mod h1:YD8tP3GAjkrDg1eZH7EGmyESg/lsYskCTPBJVb9jqSc= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/host/framing.go b/host/framing.go new file mode 100644 index 00000000..18953568 --- /dev/null +++ b/host/framing.go @@ -0,0 +1,46 @@ +package host + +import ( + "bufio" + "encoding/binary" + "fmt" + "io" +) + +func writeFrame(w io.Writer, payload []byte) error { + var prefix [binary.MaxVarintLen64]byte + n := binary.PutUvarint(prefix[:], uint64(len(payload))) + if err := writeAll(w, prefix[:n]); err != nil { + return err + } + return writeAll(w, payload) +} + +func readFrame(r *bufio.Reader, maximum int) ([]byte, error) { + size, err := binary.ReadUvarint(r) + if err != nil { + return nil, err + } + if maximum > 0 && size > uint64(maximum) { + return nil, fmt.Errorf("host: frame length %d exceeds %d", size, maximum) + } + payload := make([]byte, int(size)) + if _, err := io.ReadFull(r, payload); err != nil { + return nil, err + } + return payload, nil +} + +func writeAll(w io.Writer, payload []byte) error { + for len(payload) > 0 { + n, err := w.Write(payload) + if err != nil { + return err + } + if n <= 0 { + return io.ErrShortWrite + } + payload = payload[n:] + } + return nil +} diff --git a/host/framing_test.go b/host/framing_test.go new file mode 100644 index 00000000..da410b72 --- /dev/null +++ b/host/framing_test.go @@ -0,0 +1,22 @@ +package host + +import ( + "bufio" + "bytes" + "testing" +) + +func TestFrameRoundTrip(t *testing.T) { + var stream bytes.Buffer + want := []byte(`{"component":"Counter","payload":{"value":42}}`) + if err := writeFrame(&stream, want); err != nil { + t.Fatal(err) + } + got, err := readFrame(bufio.NewReader(&stream), 1024) + if err != nil { + t.Fatal(err) + } + if !bytes.Equal(got, want) { + t.Fatalf("frame = %q, want %q", got, want) + } +} diff --git a/host/server.go b/host/server.go index 5fcba822..8cf3459c 100644 --- a/host/server.go +++ b/host/server.go @@ -98,6 +98,7 @@ func NewMux(root string, opts ...MuxOption) *http.ServeMux { mux.Handle("/ws", runtime.Guard(websocket.Handler(func(ws *websocket.Conn) { wsHandler(ws, runtime) }))) + registerStreamBus(mux, runtime) return mux } @@ -121,8 +122,11 @@ func ListenAndServeTLS(addr, root string) error { if err != nil { return err } - srv := newHTTPServer(addr, loggingMiddleware(NewMux(root))) - srv.TLSConfig = &tls.Config{Certificates: []tls.Certificate{cert}} + mux := NewMux(root) + tlsConfig := &tls.Config{Certificates: []tls.Certificate{cert}} + startStreamBusHTTP3(addr, mux, tlsConfig) + srv := newHTTPServer(addr, loggingMiddleware(mux)) + srv.TLSConfig = tlsConfig logger.Info("serving HTTPS", "addr", addr) return srv.ListenAndServeTLS("", "") } @@ -134,8 +138,10 @@ func ListenAndServeTLSWithMux(addr string, mux *http.ServeMux) error { if err != nil { return err } + tlsConfig := &tls.Config{Certificates: []tls.Certificate{cert}} + startStreamBusHTTP3(addr, mux, tlsConfig) srv := newHTTPServer(addr, loggingMiddleware(mux)) - srv.TLSConfig = &tls.Config{Certificates: []tls.Certificate{cert}} + srv.TLSConfig = tlsConfig logger.Info("serving HTTPS", "addr", addr) return srv.ListenAndServeTLS("", "") } @@ -198,10 +204,12 @@ func generateSelfSignedCert() (tls.Certificate, error) { tmpl := x509.Certificate{ SerialNumber: big.NewInt(1), NotBefore: time.Now(), - NotAfter: time.Now().Add(365 * 24 * time.Hour), - KeyUsage: x509.KeyUsageKeyEncipherment | x509.KeyUsageDigitalSignature, - ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth}, - DNSNames: []string{"localhost"}, + // WebTransport accepts certificate hashes for development certificates + // only when their validity is at most two weeks. + NotAfter: time.Now().Add(13 * 24 * time.Hour), + KeyUsage: x509.KeyUsageKeyEncipherment | x509.KeyUsageDigitalSignature, + ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth}, + DNSNames: []string{"localhost"}, } der, err := x509.CreateCertificate(rand.Reader, &tmpl, &tmpl, &priv.PublicKey, priv) if err != nil { diff --git a/host/session.go b/host/session.go index 378c0a54..0eb09ce6 100644 --- a/host/session.go +++ b/host/session.go @@ -24,6 +24,7 @@ type Session struct { deliveryMu sync.Mutex outboundMu sync.Mutex connection *websocket.Conn + streamConnection *streamBusConnection connectionManaged bool resumePending bool attached bool @@ -145,6 +146,7 @@ func SuspendSession(session *Session, ttl time.Duration) { return } session.connection = nil + session.streamConnection = nil session.attached = false if ttl <= 0 || session.resumeToken == "" { session.deliveryMu.Unlock() @@ -259,6 +261,7 @@ func releaseSession(session *Session, expectedExpiry time.Time) { session.expiryTimer = nil } session.connection = nil + session.streamConnection = nil session.connectionManaged = false session.resumePending = false session.attached = false diff --git a/host/streambus_benchmark_test.go b/host/streambus_benchmark_test.go new file mode 100644 index 00000000..9610f14d --- /dev/null +++ b/host/streambus_benchmark_test.go @@ -0,0 +1,103 @@ +package host + +import ( + "context" + "encoding/json" + "fmt" + "net/http/httptest" + "strings" + "testing" + + "github.com/mirkobrombin/go-warp/v2/streambus" + "golang.org/x/net/websocket" +) + +func BenchmarkStreamBusFanout(b *testing.B) { + for _, subscribers := range []int{1, 10, 100} { + b.Run(fmt.Sprintf("subscribers-%d", subscribers), func(b *testing.B) { + bus := streambus.NewInMemory(streambus.Config{DefaultBuffer: 4096, MaxBuffer: 4096}) + defer bus.Close() + topics := make([]string, subscribers) + for i := 0; i < subscribers; i++ { + topics[i] = fmt.Sprintf("rfw/connection/%d", i) + subscription, err := bus.Subscribe(context.Background(), streambus.SubscribeOptions{ + Topic: topics[i], Buffer: 4096, Overflow: streambus.Block, + }) + if err != nil { + b.Fatal(err) + } + defer subscription.Close() + go func() { + for range subscription.Frames() { + } + }() + } + outbound := Outbound{Component: "Chart", Payload: map[string]any{"value": 42}} + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + for _, topic := range topics { + payload, err := json.Marshal(outbound) + if err != nil { + b.Fatal(err) + } + if _, err := bus.Publish(context.Background(), streambus.Frame{ + Topic: topic, Payload: payload, Reliability: streambus.Reliable, + Priority: streambus.PriorityInteractive, + }); err != nil { + b.Fatal(err) + } + } + } + }) + } +} + +func BenchmarkWebSocketFanout(b *testing.B) { + for _, subscribers := range []int{1, 10, 100} { + b.Run(fmt.Sprintf("subscribers-%d", subscribers), func(b *testing.B) { + accepted := make(chan *websocket.Conn, subscribers) + stop := make(chan struct{}) + server := httptest.NewServer(websocket.Handler(func(connection *websocket.Conn) { + accepted <- connection + <-stop + })) + defer server.Close() + serverConnections := make([]*websocket.Conn, 0, subscribers) + clientConnections := make([]*websocket.Conn, 0, subscribers) + for i := 0; i < subscribers; i++ { + client, err := websocket.Dial("ws"+strings.TrimPrefix(server.URL, "http"), "", server.URL) + if err != nil { + b.Fatal(err) + } + clientConnections = append(clientConnections, client) + serverConnections = append(serverConnections, <-accepted) + go func() { + for { + var payload []byte + if websocket.Message.Receive(client, &payload) != nil { + return + } + } + }() + } + defer func() { + for _, connection := range clientConnections { + _ = connection.Close() + } + for _, connection := range serverConnections { + ForgetConnection(connection) + } + close(stop) + }() + outbound := Outbound{Component: "Chart", Payload: map[string]any{"value": 42}} + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + for _, connection := range serverConnections { + SendOutbound(connection, outbound) + } + } + }) + } +} diff --git a/host/streambus_integration_test.go b/host/streambus_integration_test.go new file mode 100644 index 00000000..6733d75f --- /dev/null +++ b/host/streambus_integration_test.go @@ -0,0 +1,107 @@ +package host + +import ( + "bufio" + "context" + "crypto/tls" + "encoding/json" + "fmt" + "net" + "net/http" + "os" + "path/filepath" + "testing" + "time" + + "github.com/quic-go/quic-go" + "github.com/quic-go/quic-go/http3" + wt "github.com/quic-go/webtransport-go" +) + +func TestStreamBusWebTransportProtocol(t *testing.T) { + t.Setenv("RFW_TRANSPORT", "streambus") + root := t.TempDir() + if err := os.WriteFile(filepath.Join(root, "index.html"), []byte("ok"), 0o644); err != nil { + t.Fatal(err) + } + mux := NewMux(root) + value, ok := streamEndpoints.Load(mux) + if !ok { + t.Fatal("StreamBus endpoint was not registered") + } + endpoint := value.(*streamBusEndpoint) + + certificate, err := generateSelfSignedCert() + if err != nil { + t.Fatal(err) + } + serverTLS := &tls.Config{Certificates: []tls.Certificate{certificate}, NextProtos: []string{http3.NextProtoH3}} + h3 := &http3.Server{ + TLSConfig: serverTLS, Handler: mux, EnableDatagrams: true, + QUICConfig: &quic.Config{EnableDatagrams: true, EnableStreamResetPartialDelivery: true}, + } + server := &wt.Server{H3: h3} + wt.ConfigureHTTP3Server(h3) + endpoint.mu.Lock() + endpoint.server = server + endpoint.mu.Unlock() + + address, err := net.ResolveUDPAddr("udp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + packet, err := net.ListenUDP("udp", address) + if err != nil { + t.Fatal(err) + } + done := make(chan error, 1) + go func() { done <- server.Serve(packet) }() + t.Cleanup(func() { + _ = server.Close() + _ = packet.Close() + select { + case <-done: + case <-time.After(time.Second): + t.Error("WebTransport server did not stop") + } + }) + + dialer := &wt.Dialer{ + TLSClientConfig: &tls.Config{InsecureSkipVerify: true}, + QUICConfig: &quic.Config{EnableDatagrams: true, EnableStreamResetPartialDelivery: true}, + } + t.Cleanup(func() { _ = dialer.Close() }) + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + url := fmt.Sprintf("https://localhost:%d%s", packet.LocalAddr().(*net.UDPAddr).Port, streamBusPath) + response, session, err := dialer.Dial(ctx, url, nil) + if err != nil { + t.Fatal(err) + } + if response.StatusCode != http.StatusOK { + t.Fatalf("status = %d", response.StatusCode) + } + t.Cleanup(func() { _ = session.CloseWithError(0, "") }) + stream, err := session.OpenStreamSync(ctx) + if err != nil { + t.Fatal(err) + } + inbound, err := json.Marshal(Inbound{Component: "missing", Sequence: 1}) + if err != nil { + t.Fatal(err) + } + if err := writeFrame(stream, inbound); err != nil { + t.Fatal(err) + } + payload, err := readFrame(bufio.NewReader(stream), 1<<20) + if err != nil { + t.Fatal(err) + } + var outbound Outbound + if err := json.Unmarshal(payload, &outbound); err != nil { + t.Fatal(err) + } + if outbound.Control != "ack" || outbound.Session == "" || outbound.Sequence == 0 { + t.Fatalf("unexpected response: %#v", outbound) + } +} diff --git a/host/streambus_server.go b/host/streambus_server.go new file mode 100644 index 00000000..f5a42e2e --- /dev/null +++ b/host/streambus_server.go @@ -0,0 +1,389 @@ +package host + +import ( + "bufio" + "context" + "crypto/sha256" + "crypto/tls" + "encoding/json" + "errors" + "fmt" + "io" + "log" + "net" + "net/http" + "strings" + "sync" + "sync/atomic" + "time" + + "github.com/mirkobrombin/go-warp/v2/streambus" + "github.com/quic-go/quic-go" + "github.com/quic-go/quic-go/http3" + wt "github.com/quic-go/webtransport-go" +) + +const streamBusPath = "/streambus" + +var ( + streamEndpoints sync.Map + streamClientsMu sync.RWMutex + streamClients = make(map[string]map[*streamBusConnection]*Session) + streamID atomic.Uint64 + streamCertHash atomic.Value + streamHTTP3Port atomic.Value +) + +type streamBusEndpoint struct { + runtime *WSRuntime + bus *streambus.InMemory + mu sync.RWMutex + server *wt.Server +} + +func newStreamBusEndpoint(runtime *WSRuntime) *streamBusEndpoint { + maximum := DefaultSSCLimits().MaxMessageBytes + if runtime != nil && runtime.limits.MaxMessageBytes > 0 { + maximum = runtime.limits.MaxMessageBytes + } + return &streamBusEndpoint{ + runtime: runtime, + bus: streambus.NewInMemory(streambus.Config{ + DefaultBuffer: 256, MaxBuffer: 4096, ReplayCapacity: 256, + MaxPayloadBytes: maximum, + }), + } +} + +func registerStreamBus(mux *http.ServeMux, runtime *WSRuntime) { + if !streamBusEnabled() { + return + } + endpoint := newStreamBusEndpoint(runtime) + mux.Handle(streamBusPath, runtime.Guard(endpoint)) + mux.HandleFunc("/__rfw/streambus-config", func(w http.ResponseWriter, _ *http.Request) { + hash, _ := streamCertHash.Load().(string) + port, _ := streamHTTP3Port.Load().(string) + if hash == "" || port == "" { + http.Error(w, "streambus certificate is not ready", http.StatusServiceUnavailable) + return + } + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(map[string]string{"certificateHash": hash, "port": port}) + }) + streamEndpoints.Store(mux, endpoint) +} + +func (e *streamBusEndpoint) ServeHTTP(w http.ResponseWriter, r *http.Request) { + e.mu.RLock() + server := e.server + e.mu.RUnlock() + if server == nil { + http.Error(w, "streambus requires HTTP/3", http.StatusUpgradeRequired) + return + } + session, err := server.Upgrade(w, r) + if err != nil { + return + } + if err := e.serveSession(r, session); err != nil && !errors.Is(err, context.Canceled) && !errors.Is(err, io.EOF) { + log.Printf("streambus: %v", err) + } +} + +func (e *streamBusEndpoint) serveSession(request *http.Request, transport *wt.Session) error { + if !e.runtime.AcquireConnection() { + _ = transport.CloseWithError(1, "connection limit reached") + return errors.New("streambus: connection limit reached") + } + defer e.runtime.ReleaseConnection() + stream, err := transport.AcceptStream(transport.Context()) + if err != nil { + return err + } + connection, err := newStreamBusConnection(transport.Context(), request, transport, stream, e.bus, e.runtime.limits.MaxMessageBytes) + if err != nil { + return err + } + defer connection.Close() + return streamBusProtocolLoop(connection, e.runtime) +} + +type streamBusConnection struct { + request *http.Request + transport *wt.Session + stream *wt.Stream + reader *bufio.Reader + bus *streambus.InMemory + topic string + subscription *streambus.Subscription + maximum int + writeMu sync.Mutex + closeOnce sync.Once + done chan struct{} +} + +func newStreamBusConnection(ctx context.Context, request *http.Request, transport *wt.Session, stream *wt.Stream, bus *streambus.InMemory, maximum int) (*streamBusConnection, error) { + topic := fmt.Sprintf("rfw/connection/%d", streamID.Add(1)) + subscription, err := bus.Subscribe(ctx, streambus.SubscribeOptions{Topic: topic, Buffer: 256, Overflow: streambus.Block}) + if err != nil { + return nil, err + } + connection := &streamBusConnection{ + request: request, transport: transport, stream: stream, reader: bufio.NewReader(stream), + bus: bus, topic: topic, subscription: subscription, maximum: maximum, done: make(chan struct{}), + } + go connection.writeLoop() + return connection, nil +} + +func (c *streamBusConnection) Receive(message *Inbound) error { + payload, err := readFrame(c.reader, c.maximum) + if err != nil { + return err + } + return json.Unmarshal(payload, message) +} + +func (c *streamBusConnection) Send(out Outbound) error { + payload, err := json.Marshal(out) + if err != nil { + return err + } + ctx, cancel := context.WithTimeout(c.transport.Context(), 5*time.Second) + defer cancel() + _, err = c.bus.Publish(ctx, streambus.Frame{ + Topic: c.topic, Payload: payload, Reliability: streambus.Reliable, + Priority: streambus.PriorityInteractive, + }) + return err +} + +func (c *streamBusConnection) writeLoop() { + defer close(c.done) + for frame := range c.subscription.Frames() { + c.writeMu.Lock() + err := writeFrame(c.stream, frame.Payload) + c.writeMu.Unlock() + if err != nil { + _ = c.transport.CloseWithError(2, "write failed") + return + } + } +} + +func (c *streamBusConnection) Close() { + c.closeOnce.Do(func() { + _ = c.transport.CloseWithError(0, "") + _ = c.subscription.Close() + c.writeMu.Lock() + _ = c.stream.Close() + c.writeMu.Unlock() + }) +} + +func streamBusProtocolLoop(connection *streamBusConnection, runtime *WSRuntime) error { + var session *Session + var subscribed []string + defer func() { + streamClientsMu.Lock() + for _, name := range subscribed { + if set := streamClients[name]; set != nil { + delete(set, connection) + if len(set) == 0 { + delete(streamClients, name) + } + } + } + streamClientsMu.Unlock() + SuspendSession(session, runtime.ResumeTTL()) + }() + + for { + var message Inbound + if err := connection.Receive(&message); err != nil { + return err + } + if session == nil { + var resumed bool + var err error + session, resumed, err = runtime.OpenSession(connection.request, message.ResumeToken) + if err != nil { + _ = connection.Send(Outbound{Error: NewActionError("session_rejected", "session rejected")}) + return err + } + if !bindStreamBusConnection(connection, session, true) { + _ = connection.Send(Outbound{Error: NewActionError("session_rejected", "session already has an active connection")}) + return errors.New("streambus: session already has an active connection") + } + if resumed { + replayStreamBus(connection, session, message.Ack) + } else if message.ResumeToken != "" { + sendStreamBusSession(connection, session, Outbound{Control: "resume_rejected", Error: NewActionError("resume_rejected", "session could not be resumed")}) + } + } + session.Acknowledge(message.Ack) + if message.Control == "ping" { + sendStreamBusSession(connection, session, Outbound{Control: "pong"}) + continue + } + if err := session.AcceptInbound(message.Sequence); err != nil { + if errors.Is(err, ErrDuplicateMessage) { + continue + } + sendStreamBusSession(connection, session, Outbound{ID: message.ID, Action: message.Action, Error: NewActionError("sequence_gap", "client message sequence gap")}) + continue + } + if !session.AllowMessage(runtime.MessagesPerMinute()) { + sendStreamBusSession(connection, session, Outbound{ID: message.ID, Action: message.Action, Error: NewActionError("rate_limited", "message rate limit exceeded")}) + continue + } + authorizeCtx, cancel := runtime.HandlerContext(context.Background()) + authorizeErr := runtime.Authorize(authorizeCtx, session, message) + cancel() + if authorizeErr != nil { + sendStreamBusSession(connection, session, Outbound{Component: message.Component, Action: message.Action, ID: message.ID, Error: NewActionError("forbidden", "message forbidden")}) + continue + } + if message.Action != "" { + payload, actionErr := runtime.DispatchAction(context.Background(), session, message) + sendStreamBusSession(connection, session, Outbound{Action: message.Action, ID: message.ID, Payload: payload, Error: actionErr}) + continue + } + if message.Component != "" && message.Payload != nil && message.Payload["unsubscribe"] == true { + streamClientsMu.Lock() + if set := streamClients[message.Component]; set != nil { + delete(set, connection) + if len(set) == 0 { + delete(streamClients, message.Component) + } + } + for index, name := range subscribed { + if name == message.Component { + subscribed = append(subscribed[:index], subscribed[index+1:]...) + break + } + } + streamClientsMu.Unlock() + sendStreamBusSession(connection, session, Outbound{Component: message.Component, Control: "unsubscribed"}) + continue + } + if component, ok := Get(message.Component); ok { + streamClientsMu.Lock() + if streamClients[message.Component] == nil { + streamClients[message.Component] = make(map[*streamBusConnection]*Session) + } + if _, tracked := streamClients[message.Component][connection]; !tracked { + streamClients[message.Component][connection] = session + subscribed = append(subscribed, message.Component) + } + streamClientsMu.Unlock() + response := component.HandleWithSession(session, message.Payload) + if response != nil { + switch value := response.(type) { + case *InitSnapshot: + if value != nil { + sendStreamBusSession(connection, session, Outbound{Component: message.Component, ID: message.ID, Payload: map[string]any{"initSnapshot": value}}) + } + case InitSnapshot: + sendStreamBusSession(connection, session, Outbound{Component: message.Component, ID: message.ID, Payload: map[string]any{"initSnapshot": value}}) + default: + sendStreamBusSession(connection, session, Outbound{Component: message.Component, ID: message.ID, Payload: response}) + } + continue + } + if message.Payload != nil && message.Payload["init"] == true { + sendStreamBusSession(connection, session, Outbound{Component: message.Component, Payload: map[string]any{"session": session.ID()}}) + continue + } + } + sendStreamBusSession(connection, session, Outbound{Control: "ack"}) + } +} + +func bindStreamBusConnection(connection *streamBusConnection, session *Session, handoff bool) bool { + if session == nil || connection == nil { + return false + } + session.outboundMu.Lock() + defer session.outboundMu.Unlock() + session.deliveryMu.Lock() + active := session.attached && !session.released + resumePending := session.resumePending + if active && resumePending && handoff { + session.resumePending = false + } + session.deliveryMu.Unlock() + if !active || (resumePending && !handoff) { + return false + } + if session.streamConnection == connection { + return true + } + if session.streamConnection != nil || session.connection != nil || (session.connectionManaged && !handoff) { + return false + } + session.streamConnection = connection + session.connection = nil + session.connectionManaged = true + return true +} + +func sendStreamBusSession(connection *streamBusConnection, session *Session, out Outbound) { + if session == nil { + return + } + session.outboundMu.Lock() + defer session.outboundMu.Unlock() + if session.streamConnection != connection { + return + } + _ = connection.Send(session.PrepareOutbound(out)) +} + +func replayStreamBus(connection *streamBusConnection, session *Session, acknowledged uint64) { + session.outboundMu.Lock() + defer session.outboundMu.Unlock() + if session.streamConnection != connection { + return + } + messages, err := session.ReplayAfter(acknowledged) + if err != nil { + _ = connection.Send(session.PrepareOutbound(Outbound{Error: NewActionError("resync_required", "message history is no longer available")})) + return + } + for _, message := range messages { + _ = connection.Send(message) + } +} + +func startStreamBusHTTP3(addr string, mux *http.ServeMux, tlsConfig *tls.Config) { + value, ok := streamEndpoints.Load(mux) + if !ok { + return + } + endpoint := value.(*streamBusEndpoint) + port := strings.TrimPrefix(addr, ":") + if _, parsedPort, err := net.SplitHostPort(addr); err == nil { + port = parsedPort + } + streamHTTP3Port.Store(port) + if len(tlsConfig.Certificates) > 0 && len(tlsConfig.Certificates[0].Certificate) > 0 { + hash := sha256.Sum256(tlsConfig.Certificates[0].Certificate[0]) + streamCertHash.Store(fmt.Sprintf("%x", hash[:])) + } + h3 := &http3.Server{ + Addr: addr, Handler: mux, TLSConfig: tlsConfig.Clone(), EnableDatagrams: true, + QUICConfig: &quic.Config{EnableDatagrams: true, EnableStreamResetPartialDelivery: true}, + } + server := &wt.Server{H3: h3} + wt.ConfigureHTTP3Server(h3) + endpoint.mu.Lock() + endpoint.server = server + endpoint.mu.Unlock() + go func() { + if err := server.ListenAndServe(); err != nil { + log.Printf("streambus HTTP/3 server: %v", err) + } + }() +} diff --git a/host/transport.go b/host/transport.go new file mode 100644 index 00000000..26162ac2 --- /dev/null +++ b/host/transport.go @@ -0,0 +1,53 @@ +package host + +import ( + "encoding/json" + "os" + "strings" +) + +// Transport identifies the browser-to-host transport used by SSC. +type Transport string + +const ( + TransportWebSocket Transport = "websocket" + TransportStreamBus Transport = "streambus" + TransportAuto Transport = "auto" +) + +// ResolveTransport reads RFW_TRANSPORT first, then the top-level transport +// key in rfw.json. Unknown and empty values preserve the WebSocket default. +func ResolveTransport() Transport { + if mode := normalizeTransport(os.Getenv("RFW_TRANSPORT")); mode != "" { + return mode + } + var manifest struct { + Transport string `json:"transport"` + } + if data, err := os.ReadFile("rfw.json"); err == nil { + if json.Unmarshal(data, &manifest) == nil { + if mode := normalizeTransport(manifest.Transport); mode != "" { + return mode + } + } + } + return TransportWebSocket +} + +func normalizeTransport(value string) Transport { + switch strings.ToLower(strings.TrimSpace(value)) { + case string(TransportWebSocket), "ws": + return TransportWebSocket + case string(TransportStreamBus), "webtransport", "warp-streambus": + return TransportStreamBus + case string(TransportAuto): + return TransportAuto + default: + return "" + } +} + +func streamBusEnabled() bool { + mode := ResolveTransport() + return mode == TransportStreamBus || mode == TransportAuto +} diff --git a/host/transport_test.go b/host/transport_test.go new file mode 100644 index 00000000..e8a489c6 --- /dev/null +++ b/host/transport_test.go @@ -0,0 +1,48 @@ +package host + +import ( + "os" + "path/filepath" + "testing" +) + +func TestResolveTransportPrecedence(t *testing.T) { + dir := t.TempDir() + previous, err := os.Getwd() + if err != nil { + t.Fatal(err) + } + if err := os.Chdir(dir); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = os.Chdir(previous) }) + + if got := ResolveTransport(); got != TransportWebSocket { + t.Fatalf("default = %q", got) + } + if err := os.WriteFile(filepath.Join(dir, "rfw.json"), []byte(`{"transport":"streambus"}`), 0o644); err != nil { + t.Fatal(err) + } + if got := ResolveTransport(); got != TransportStreamBus { + t.Fatalf("manifest = %q", got) + } + t.Setenv("RFW_TRANSPORT", "ws") + if got := ResolveTransport(); got != TransportWebSocket { + t.Fatalf("environment = %q", got) + } +} + +func TestResolveTransportAliases(t *testing.T) { + for value, want := range map[string]Transport{ + "websocket": TransportWebSocket, + "WS": TransportWebSocket, + "streambus": TransportStreamBus, + "webtransport": TransportStreamBus, + "warp-streambus": TransportStreamBus, + "auto": TransportAuto, + } { + if got := normalizeTransport(value); got != want { + t.Errorf("normalizeTransport(%q) = %q, want %q", value, got, want) + } + } +} diff --git a/host/websocket.go b/host/websocket.go index 826ddf5f..910caffa 100644 --- a/host/websocket.go +++ b/host/websocket.go @@ -388,6 +388,25 @@ func Broadcast(name string, payload any, opts ...BroadcastOption) { } SendSessionOutbound(t.ws, t.session, Outbound{Component: name, Payload: payload}) } + + streamClientsMu.RLock() + streamTargets := make([]struct { + connection *streamBusConnection + session *Session + }, 0, len(streamClients[name])) + for connection, session := range streamClients[name] { + streamTargets = append(streamTargets, struct { + connection *streamBusConnection + session *Session + }{connection: connection, session: session}) + } + streamClientsMu.RUnlock() + for _, target := range streamTargets { + if options.Session != "" && target.session.ID() != options.Session { + continue + } + sendStreamBusSession(target.connection, target.session, Outbound{Component: name, Payload: payload}) + } } // DispatchAction executes a typed action within the configured handler deadline. @@ -530,10 +549,11 @@ func sessionAcceptsConnection(session *Session, ws *websocket.Conn, handoff bool if session.connection == ws { return true, false } - if session.connection != nil || (session.connectionManaged && !handoff) { + if session.connection != nil || session.streamConnection != nil || (session.connectionManaged && !handoff) { return false, false } session.connection = ws + session.streamConnection = nil return true, false } diff --git a/hostclient/streambus_transport_wasm.go b/hostclient/streambus_transport_wasm.go new file mode 100644 index 00000000..420747b5 --- /dev/null +++ b/hostclient/streambus_transport_wasm.go @@ -0,0 +1,268 @@ +//go:build js && wasm + +package hostclient + +import ( + "bufio" + "context" + "encoding/binary" + "encoding/hex" + "errors" + "fmt" + "io" + "net" + "net/url" + "strconv" + "strings" + stdjs "syscall/js" +) + +const ( + hostTransportWebSocket = "websocket" + hostTransportStreamBus = "streambus" + hostTransportAuto = "auto" + streamBusPath = "/streambus" +) + +func preferredHostTransport() string { + value := strings.ToLower(strings.TrimSpace(stdjs.Global().Get("RFW_TRANSPORT").String())) + switch value { + case hostTransportStreamBus, "webtransport", "warp-streambus": + return hostTransportStreamBus + case hostTransportAuto: + return hostTransportAuto + default: + return hostTransportWebSocket + } +} + +func hostStreamBusURL() string { + if configured := stdjs.Global().Get("RFW_STREAMBUS_URL"); configured.Truthy() { + return normalizeStreamBusURL(configured.String(), false) + } + return normalizeStreamBusURL(hostWSURL(), true) +} + +func normalizeStreamBusURL(raw string, incrementHTTPPort bool) string { + raw = strings.TrimSpace(raw) + if raw == "" { + return "" + } + switch { + case strings.HasPrefix(raw, "wss://"): + raw = "https://" + strings.TrimPrefix(raw, "wss://") + incrementHTTPPort = false + case strings.HasPrefix(raw, "ws://"): + raw = "https://" + strings.TrimPrefix(raw, "ws://") + case strings.HasPrefix(raw, "http://"): + raw = "https://" + strings.TrimPrefix(raw, "http://") + case strings.HasPrefix(raw, "https://"): + incrementHTTPPort = false + default: + raw = "https://" + raw + } + parsed, err := url.Parse(raw) + if err != nil || parsed.Hostname() == "" { + return "" + } + if incrementHTTPPort { + if port, err := strconv.Atoi(parsed.Port()); err == nil && port > 0 { + parsed.Host = net.JoinHostPort(parsed.Hostname(), strconv.Itoa(port+1)) + } + } + if parsed.Path == "" || parsed.Path == "/" || parsed.Path == "/ws" { + parsed.Path = streamBusPath + } + parsed.RawQuery = "" + parsed.Fragment = "" + return parsed.String() +} + +func dialStreamBus(ctx context.Context, endpoint string) (*hostConn, error) { + constructor := stdjs.Global().Get("WebTransport") + if constructor.Type() != stdjs.TypeFunction { + return nil, errors.New("hostclient: WebTransport API is unavailable") + } + config, _ := loadStreamBusConfig(ctx) + if config.port != "" { + endpoint = replaceURLPort(endpoint, config.port) + } + var transport stdjs.Value + if config.options.Type() == stdjs.TypeObject { + transport = constructor.New(endpoint, config.options) + } else { + transport = constructor.New(endpoint) + } + if _, err := awaitPromise(ctx, transport.Get("ready")); err != nil { + transport.Call("close") + return nil, err + } + stream, err := awaitPromise(ctx, transport.Call("createBidirectionalStream")) + if err != nil { + transport.Call("close") + return nil, err + } + reader := &jsReader{reader: stream.Get("readable").Call("getReader")} + writer := &jsWriter{writer: stream.Get("writable").Call("getWriter")} + buffered := bufio.NewReader(reader) + c := newHostConn() + c.send = func(data string) error { + return writeStreamBusFrame(context.Background(), writer, []byte(data)) + } + c.shutdown = func() error { + reader.reader.Call("cancel") + writer.writer.Call("close") + transport.Call("close") + return nil + } + c.openOnce.Do(func() { close(c.open) }) + go func() { + for { + payload, readErr := readStreamBusFrame(buffered, int(maxInboundMessageBytes)) + if readErr != nil { + c.fail(readErr) + return + } + c.deliver(payload) + } + }() + return c, nil +} + +type streamBusClientConfig struct { + options stdjs.Value + port string +} + +func loadStreamBusConfig(ctx context.Context) (streamBusClientConfig, error) { + fetch := stdjs.Global().Get("fetch") + if fetch.Type() != stdjs.TypeFunction { + return streamBusClientConfig{options: stdjs.Undefined()}, nil + } + response, err := awaitPromise(ctx, fetch.Invoke("/__rfw/streambus-config")) + if err != nil || !response.Get("ok").Bool() { + return streamBusClientConfig{options: stdjs.Undefined()}, err + } + config, err := awaitPromise(ctx, response.Call("json")) + if err != nil { + return streamBusClientConfig{options: stdjs.Undefined()}, err + } + hash, err := hex.DecodeString(config.Get("certificateHash").String()) + if err != nil || len(hash) != 32 { + return streamBusClientConfig{options: stdjs.Undefined()}, errors.New("hostclient: invalid StreamBus certificate hash") + } + value := stdjs.Global().Get("Uint8Array").New(len(hash)) + stdjs.CopyBytesToJS(value, hash) + descriptor := stdjs.Global().Get("Object").New() + descriptor.Set("algorithm", "sha-256") + descriptor.Set("value", value.Get("buffer")) + list := stdjs.Global().Get("Array").New() + list.Call("push", descriptor) + options := stdjs.Global().Get("Object").New() + options.Set("serverCertificateHashes", list) + return streamBusClientConfig{options: options, port: config.Get("port").String()}, nil +} + +func replaceURLPort(raw, port string) string { + parsed, err := url.Parse(raw) + if err != nil || parsed.Hostname() == "" || port == "" { + return raw + } + parsed.Host = net.JoinHostPort(parsed.Hostname(), port) + return parsed.String() +} + +func writeStreamBusFrame(ctx context.Context, writer *jsWriter, payload []byte) error { + var prefix [binary.MaxVarintLen64]byte + n := binary.PutUvarint(prefix[:], uint64(len(payload))) + if err := writer.write(ctx, prefix[:n]); err != nil { + return err + } + return writer.write(ctx, payload) +} + +func readStreamBusFrame(reader *bufio.Reader, maximum int) ([]byte, error) { + size, err := binary.ReadUvarint(reader) + if err != nil { + return nil, err + } + if maximum > 0 && size > uint64(maximum) { + return nil, fmt.Errorf("hostclient: StreamBus frame of %d bytes exceeds %d", size, maximum) + } + payload := make([]byte, int(size)) + if _, err := io.ReadFull(reader, payload); err != nil { + return nil, err + } + return payload, nil +} + +type jsReader struct { + reader stdjs.Value + buffer []byte +} + +func (r *jsReader) Read(payload []byte) (int, error) { + if len(r.buffer) == 0 { + result, err := awaitPromise(context.Background(), r.reader.Call("read")) + if err != nil { + return 0, err + } + if result.Get("done").Bool() { + return 0, io.EOF + } + value := result.Get("value") + r.buffer = make([]byte, value.Get("byteLength").Int()) + stdjs.CopyBytesToGo(r.buffer, value) + } + n := copy(payload, r.buffer) + r.buffer = r.buffer[n:] + return n, nil +} + +type jsWriter struct{ writer stdjs.Value } + +func (w *jsWriter) write(ctx context.Context, payload []byte) error { + array := stdjs.Global().Get("Uint8Array").New(len(payload)) + stdjs.CopyBytesToJS(array, payload) + _, err := awaitPromise(ctx, w.writer.Call("write", array)) + return err +} + +type promiseResult struct { + value stdjs.Value + err error +} + +func awaitPromise(ctx context.Context, promise stdjs.Value) (stdjs.Value, error) { + result := make(chan promiseResult, 1) + resolve := stdjs.FuncOf(func(_ stdjs.Value, args []stdjs.Value) any { + value := stdjs.Undefined() + if len(args) > 0 { + value = args[0] + } + result <- promiseResult{value: value} + return nil + }) + reject := stdjs.FuncOf(func(_ stdjs.Value, args []stdjs.Value) any { + message := "promise rejected" + if len(args) > 0 { + message = args[0].String() + } + result <- promiseResult{err: errors.New(message)} + return nil + }) + promise.Call("then", resolve).Call("catch", reject) + select { + case settled := <-result: + resolve.Release() + reject.Release() + return settled.value, settled.err + case <-ctx.Done(): + go func() { + <-result + resolve.Release() + reject.Release() + }() + return stdjs.Undefined(), ctx.Err() + } +} diff --git a/hostclient/streambus_transport_wasm_test.go b/hostclient/streambus_transport_wasm_test.go new file mode 100644 index 00000000..120cc4e7 --- /dev/null +++ b/hostclient/streambus_transport_wasm_test.go @@ -0,0 +1,38 @@ +//go:build js && wasm + +package hostclient + +import ( + stdjs "syscall/js" + "testing" +) + +func TestStreamBusURLUsesAdvertisedHTTP3Port(t *testing.T) { + got := replaceURLPort("https://localhost:8081/streambus", "8083") + if got != "https://localhost:8083/streambus" { + t.Fatalf("URL = %q", got) + } +} + +func TestNormalizeStreamBusURLFromDevelopmentWebSocket(t *testing.T) { + got := normalizeStreamBusURL("ws://localhost:8080/ws", true) + if got != "https://localhost:8081/streambus" { + t.Fatalf("URL = %q", got) + } +} + +func TestPreferredHostTransportReadsGeneratedConfig(t *testing.T) { + global := stdjs.Global() + previous := global.Get("RFW_TRANSPORT") + global.Set("RFW_TRANSPORT", "streambus") + t.Cleanup(func() { + if previous.Type() == stdjs.TypeUndefined { + global.Delete("RFW_TRANSPORT") + } else { + global.Set("RFW_TRANSPORT", previous) + } + }) + if got := preferredHostTransport(); got != hostTransportStreamBus { + t.Fatalf("transport = %q", got) + } +} diff --git a/hostclient/transport_selector.go b/hostclient/transport_selector.go index 64e1a20d..4974a3eb 100644 --- a/hostclient/transport_selector.go +++ b/hostclient/transport_selector.go @@ -30,6 +30,14 @@ var nativeConnectionSequence atomic.Uint64 func dial(ctx context.Context, url string) (*hostConn, error) { switch sscTransport() { case sscTransportBrowser: + mode := preferredHostTransport() + if mode == hostTransportStreamBus || mode == hostTransportAuto { + if connection, err := dialStreamBus(ctx, hostStreamBusURL()); err == nil { + return connection, nil + } else if debug { + fmt.Printf("hostclient: StreamBus unavailable, falling back to WebSocket: %v\n", err) + } + } return dialBrowser(ctx, url) case sscTransportCapacitor: return dialCapacitor(ctx, url) From 66cf3211fef4df40e526afc7f49238525fce7c4c Mon Sep 17 00:00:00 2001 From: mirkobrombin Date: Wed, 2 Sep 2026 12:31:11 +0200 Subject: [PATCH 2/3] fix: bump quic-go dependencies for CVE fixes --- go.mod | 4 ++-- go.sum | 10 ++++++---- 2 files changed, 8 insertions(+), 6 deletions(-) diff --git a/go.mod b/go.mod index 40a6385b..6244fbd9 100644 --- a/go.mod +++ b/go.mod @@ -13,8 +13,8 @@ require ( github.com/mirkobrombin/go-signal/v2 v2.0.0 github.com/mirkobrombin/go-warp/v2 v2.1.0 github.com/pkg/browser v0.0.0-20240102092130-5ac0b6a4141c - github.com/quic-go/quic-go v0.59.0 - github.com/quic-go/webtransport-go v0.10.0 + github.com/quic-go/quic-go v0.60.0 + github.com/quic-go/webtransport-go v0.11.1 github.com/russross/blackfriday/v2 v2.1.0 github.com/tdewolff/minify/v2 v2.24.3 golang.org/x/net v0.58.0 diff --git a/go.sum b/go.sum index 61699c3f..3ecf71cb 100644 --- a/go.sum +++ b/go.sum @@ -28,12 +28,14 @@ github.com/pkg/browser v0.0.0-20240102092130-5ac0b6a4141c h1:+mdjkGKdHQG3305AYmd github.com/pkg/browser v0.0.0-20240102092130-5ac0b6a4141c/go.mod h1:7rwL4CYBLnjLxUqIJNnCWiEdr3bn6IUYi15bNlnbCCU= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/quic-go/go-ossfuzz-seeds v0.1.0 h1:APacT+iIaNF6fd8AGEiN3bT/Jtkd2jz4v4TzM7MFjy0= +github.com/quic-go/go-ossfuzz-seeds v0.1.0/go.mod h1:3IOHRbJIc+L6YKMwfDtJAM9Vj9k0YY4muhuyUYk5tbk= github.com/quic-go/qpack v0.6.0 h1:g7W+BMYynC1LbYLSqRt8PBg5Tgwxn214ZZR34VIOjz8= github.com/quic-go/qpack v0.6.0/go.mod h1:lUpLKChi8njB4ty2bFLX2x4gzDqXwUpaO1DP9qMDZII= -github.com/quic-go/quic-go v0.59.0 h1:OLJkp1Mlm/aS7dpKgTc6cnpynnD2Xg7C1pwL6vy/SAw= -github.com/quic-go/quic-go v0.59.0/go.mod h1:upnsH4Ju1YkqpLXC305eW3yDZ4NfnNbmQRCMWS58IKU= -github.com/quic-go/webtransport-go v0.10.0 h1:LqXXPOXuETY5Xe8ITdGisBzTYmUOy5eSj+9n4hLTjHI= -github.com/quic-go/webtransport-go v0.10.0/go.mod h1:LeGIXr5BQKE3UsynwVBeQrU1TPrbh73MGoC6jd+V7ow= +github.com/quic-go/quic-go v0.60.0 h1:xcQioE8OM66UQLeUMHltK1CCcOu3JbVB4JAQdDQSB+0= +github.com/quic-go/quic-go v0.60.0/go.mod h1:wpKpjmPpftl30sL6pFh7REVpjbcCVy4zt2vDyK1TuJk= +github.com/quic-go/webtransport-go v0.11.1 h1:rrFQMO+7/52ZDJ04fsrjIaWqn6q1z1MYo9iVFq6JtbA= +github.com/quic-go/webtransport-go v0.11.1/go.mod h1:SHgEzUFVyj+9WUSuGB1P6Zd351Pww2leWV3SwlTovkA= github.com/russross/blackfriday/v2 v2.1.0 h1:JIOH55/0cWyOuilr9/qlrm0BSXldqnqwMsf35Ld67mk= github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM= github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= From 7ecf37763f55a94bde740779cd4287c0193f433e Mon Sep 17 00:00:00 2001 From: mirkobrombin Date: Wed, 2 Sep 2026 12:50:46 +0200 Subject: [PATCH 3/3] fix: address CI failures --- .github/workflows/test-build-release.yml | 4 ++-- host/framing.go | 5 ++++- host/framing_test.go | 9 +++++++++ host/streambus_benchmark_test.go | 5 +++-- host/streambus_integration_test.go | 12 ++++++++++-- host/transport.go | 5 ++++- host/transport_test.go | 6 +++++- hostclient/streambus_transport_wasm.go | 5 ++++- 8 files changed, 41 insertions(+), 10 deletions(-) diff --git a/.github/workflows/test-build-release.yml b/.github/workflows/test-build-release.yml index b0133294..c641306c 100644 --- a/.github/workflows/test-build-release.yml +++ b/.github/workflows/test-build-release.yml @@ -54,7 +54,7 @@ jobs: - name: Set up Go uses: actions/setup-go@v5 with: - go-version: 1.25 + go-version: 1.26 - name: Build for js/wasm run: GOOS=js GOARCH=wasm go build ./... @@ -63,7 +63,7 @@ jobs: run: GOOS=js GOARCH=wasm go vet ./... - name: Install wasmbrowsertest - run: go install github.com/agnivade/wasmbrowsertest@v0.11.0 + run: go install github.com/agnivade/wasmbrowsertest@ff350cc7eb281242c0572c9e068246c0916d29ec - name: Allow Chrome sandbox (Ubuntu 24 AppArmor) run: sudo sysctl -w kernel.apparmor_restrict_unprivileged_userns=0 diff --git a/host/framing.go b/host/framing.go index 18953568..c8dca471 100644 --- a/host/framing.go +++ b/host/framing.go @@ -24,7 +24,10 @@ func readFrame(r *bufio.Reader, maximum int) ([]byte, error) { if maximum > 0 && size > uint64(maximum) { return nil, fmt.Errorf("host: frame length %d exceeds %d", size, maximum) } - payload := make([]byte, int(size)) + if size > uint64(^uint(0)>>1) { + return nil, fmt.Errorf("host: frame length %d exceeds platform limit", size) + } + payload := make([]byte, size) if _, err := io.ReadFull(r, payload); err != nil { return nil, err } diff --git a/host/framing_test.go b/host/framing_test.go index da410b72..2c165e0d 100644 --- a/host/framing_test.go +++ b/host/framing_test.go @@ -3,6 +3,7 @@ package host import ( "bufio" "bytes" + "encoding/binary" "testing" ) @@ -20,3 +21,11 @@ func TestFrameRoundTrip(t *testing.T) { t.Fatalf("frame = %q, want %q", got, want) } } + +func TestFrameRejectsPlatformOverflow(t *testing.T) { + var prefix [binary.MaxVarintLen64]byte + n := binary.PutUvarint(prefix[:], uint64(^uint(0)>>1)+1) + if _, err := readFrame(bufio.NewReader(bytes.NewReader(prefix[:n])), 0); err == nil { + t.Fatal("readFrame accepted a length larger than int") + } +} diff --git a/host/streambus_benchmark_test.go b/host/streambus_benchmark_test.go index 9610f14d..542dacce 100644 --- a/host/streambus_benchmark_test.go +++ b/host/streambus_benchmark_test.go @@ -16,7 +16,7 @@ func BenchmarkStreamBusFanout(b *testing.B) { for _, subscribers := range []int{1, 10, 100} { b.Run(fmt.Sprintf("subscribers-%d", subscribers), func(b *testing.B) { bus := streambus.NewInMemory(streambus.Config{DefaultBuffer: 4096, MaxBuffer: 4096}) - defer bus.Close() + defer func() { _ = bus.Close() }() topics := make([]string, subscribers) for i := 0; i < subscribers; i++ { topics[i] = fmt.Sprintf("rfw/connection/%d", i) @@ -26,9 +26,10 @@ func BenchmarkStreamBusFanout(b *testing.B) { if err != nil { b.Fatal(err) } - defer subscription.Close() + defer func() { _ = subscription.Close() }() go func() { for range subscription.Frames() { + continue } }() } diff --git a/host/streambus_integration_test.go b/host/streambus_integration_test.go index 6733d75f..66193da2 100644 --- a/host/streambus_integration_test.go +++ b/host/streambus_integration_test.go @@ -4,6 +4,7 @@ import ( "bufio" "context" "crypto/tls" + "crypto/x509" "encoding/json" "fmt" "net" @@ -21,7 +22,7 @@ import ( func TestStreamBusWebTransportProtocol(t *testing.T) { t.Setenv("RFW_TRANSPORT", "streambus") root := t.TempDir() - if err := os.WriteFile(filepath.Join(root, "index.html"), []byte("ok"), 0o644); err != nil { + if err := os.WriteFile(filepath.Join(root, "index.html"), []byte("ok"), 0o600); err != nil { t.Fatal(err) } mux := NewMux(root) @@ -35,6 +36,12 @@ func TestStreamBusWebTransportProtocol(t *testing.T) { if err != nil { t.Fatal(err) } + leaf, err := x509.ParseCertificate(certificate.Certificate[0]) + if err != nil { + t.Fatal(err) + } + roots := x509.NewCertPool() + roots.AddCert(leaf) serverTLS := &tls.Config{Certificates: []tls.Certificate{certificate}, NextProtos: []string{http3.NextProtoH3}} h3 := &http3.Server{ TLSConfig: serverTLS, Handler: mux, EnableDatagrams: true, @@ -67,7 +74,7 @@ func TestStreamBusWebTransportProtocol(t *testing.T) { }) dialer := &wt.Dialer{ - TLSClientConfig: &tls.Config{InsecureSkipVerify: true}, + TLSClientConfig: &tls.Config{RootCAs: roots, ServerName: "localhost"}, QUICConfig: &quic.Config{EnableDatagrams: true, EnableStreamResetPartialDelivery: true}, } t.Cleanup(func() { _ = dialer.Close() }) @@ -78,6 +85,7 @@ func TestStreamBusWebTransportProtocol(t *testing.T) { if err != nil { t.Fatal(err) } + defer func() { _ = response.Body.Close() }() if response.StatusCode != http.StatusOK { t.Fatalf("status = %d", response.StatusCode) } diff --git a/host/transport.go b/host/transport.go index 26162ac2..8c7a4904 100644 --- a/host/transport.go +++ b/host/transport.go @@ -10,9 +10,12 @@ import ( type Transport string const ( + // TransportWebSocket uses the existing SSC WebSocket connection. TransportWebSocket Transport = "websocket" + // TransportStreamBus uses WebTransport for StreamBus frames. TransportStreamBus Transport = "streambus" - TransportAuto Transport = "auto" + // TransportAuto enables StreamBus alongside the WebSocket default. + TransportAuto Transport = "auto" ) // ResolveTransport reads RFW_TRANSPORT first, then the top-level transport diff --git a/host/transport_test.go b/host/transport_test.go index e8a489c6..0858f1e6 100644 --- a/host/transport_test.go +++ b/host/transport_test.go @@ -3,10 +3,14 @@ package host import ( "os" "path/filepath" + "runtime" "testing" ) func TestResolveTransportPrecedence(t *testing.T) { + if runtime.GOOS == "js" { + t.Skip("os.Chdir is not implemented on js") + } dir := t.TempDir() previous, err := os.Getwd() if err != nil { @@ -20,7 +24,7 @@ func TestResolveTransportPrecedence(t *testing.T) { if got := ResolveTransport(); got != TransportWebSocket { t.Fatalf("default = %q", got) } - if err := os.WriteFile(filepath.Join(dir, "rfw.json"), []byte(`{"transport":"streambus"}`), 0o644); err != nil { + if err := os.WriteFile(filepath.Join(dir, "rfw.json"), []byte(`{"transport":"streambus"}`), 0o600); err != nil { t.Fatal(err) } if got := ResolveTransport(); got != TransportStreamBus { diff --git a/hostclient/streambus_transport_wasm.go b/hostclient/streambus_transport_wasm.go index 420747b5..d221f818 100644 --- a/hostclient/streambus_transport_wasm.go +++ b/hostclient/streambus_transport_wasm.go @@ -189,7 +189,10 @@ func readStreamBusFrame(reader *bufio.Reader, maximum int) ([]byte, error) { if maximum > 0 && size > uint64(maximum) { return nil, fmt.Errorf("hostclient: StreamBus frame of %d bytes exceeds %d", size, maximum) } - payload := make([]byte, int(size)) + if size > uint64(^uint(0)>>1) { + return nil, fmt.Errorf("hostclient: StreamBus frame of %d bytes exceeds platform limit", size) + } + payload := make([]byte, size) if _, err := io.ReadFull(reader, payload); err != nil { return nil, err }