From 8b4ea55788421f43ea8cdaa7fe335fc8b444c340 Mon Sep 17 00:00:00 2001 From: clover caruso Date: Mon, 5 Oct 2026 21:44:50 -0700 Subject: [PATCH] fix: reconnect browser Live Share after failed sends Retire rejected or partial sends before any later encrypted frame can reuse the socket, invalidate retained peer lines, and exercise the JavaScript boundary in the local gate. fixes #93 Assisted-by: gpt-6.1-sol --- crates/notebook/src/live/web.js | 12 +++-- crates/notebook/src/live/web.rs | 44 ++++++++++++---- tools/TESTING.md | 1 + tools/ci.py | 4 ++ tools/web/test_live.mjs | 90 +++++++++++++++++++++++++++++++++ 5 files changed, 138 insertions(+), 13 deletions(-) create mode 100644 tools/web/test_live.mjs diff --git a/crates/notebook/src/live/web.js b/crates/notebook/src/live/web.js index 603b99508a9fa378e1901557de4d30434601b17e..2a1842bb95ef7d17f949a188263cfac06697dd09 100644 --- a/crates/notebook/src/live/web.js +++ b/crates/notebook/src/live/web.js @@ -7,6 +7,7 @@ export function liveConnect(url, message, closed) { socket.binaryType = 'arraybuffer'; socket.onmessage = event => message(event.data); socket.onclose = () => { + if (!sockets.has(id)) return; liveClose(id); closed(); }; @@ -16,9 +17,14 @@ export function liveConnect(url, message, closed) { export function liveSend(id, bytes) { const socket = sockets.get(id); - if (!socket || socket.readyState !== WebSocket.OPEN) throw new Error('Relay disconnected'); - if (socket.bufferedAmount > 1048576) throw new Error('Relay cannot keep up'); - socket.send(bytes); + try { + if (!socket || socket.readyState !== WebSocket.OPEN) throw new Error('Relay disconnected'); + if (socket.bufferedAmount > 1048576) throw new Error('Relay cannot keep up'); + socket.send(bytes); + } catch (error) { + socket?.onclose?.(); + throw error; + } } export function liveClose(id) { diff --git a/crates/notebook/src/live/web.rs b/crates/notebook/src/live/web.rs index f2c1c6927c7d943745a398195ccb3d0e49fb5b98..a0b6201e200c5d69263fb2d8543fd2e8bcbf8442 100644 --- a/crates/notebook/src/live/web.rs +++ b/crates/notebook/src/live/web.rs @@ -80,12 +80,17 @@ impl Line { pub fn send(&self, kind: u16, body: &impl Encode<()>) -> io::Result<()> { let shared = self.shared.upgrade().ok_or(io::ErrorKind::NotConnected)?; let mut bytes = Vec::new(); - self.sealer + let sealed = self + .sealer .lock() .unwrap() .as_mut() .ok_or(io::ErrorKind::NotConnected)? - .send(&mut bytes, kind, body)?; + .send(&mut bytes, kind, body); + if let Err(error) = sealed { + shared.disconnected(); + return Err(error); + } shared.stream(self.slot, &bytes) } @@ -321,11 +326,18 @@ impl Shared { fn disconnected(self: &Arc) { let peers = { let mut state = self.state.lock().unwrap(); - close(state.socket); + let socket = std::mem::replace(&mut state.socket, 0); + if socket == 0 { + return; + } + close(socket); let peers: Vec<_> = state .streams .values() - .filter_map(|s| s.peer.as_ref().map(|p| p.hello.clone())) + .filter_map(|s| { + *s.line.sealer.lock().unwrap() = None; + s.peer.as_ref().map(|p| p.hello.clone()) + }) .collect(); state.streams.clear(); state.members.clear(); @@ -358,7 +370,12 @@ impl Shared { Ok(()) } - fn group(&self, kind: u16, body: &impl Encode<()>, to: Option) -> io::Result<()> { + fn group( + self: &Arc, + kind: u16, + body: &impl Encode<()>, + to: Option, + ) -> io::Result<()> { let body = minicbor::to_vec(body).map_err(io::Error::other)?; let (socket, sealed) = { let mut state = self.state.lock().unwrap(); @@ -366,9 +383,16 @@ impl Shared { .sealer .as_mut() .ok_or(io::ErrorKind::NotConnected)? - .seal(kind, &body, to.is_none())?; + .seal(kind, &body, to.is_none()); (state.socket, sealed) }; + let sealed = match sealed { + Ok(sealed) => sealed, + Err(error) => { + self.disconnected(); + return Err(error); + } + }; let mut bytes = match to { Some(slot) => [&(::relay::GROUP | 1).to_be_bytes()[..], &slot.to_be_bytes()].concat(), None => ::relay::BROADCAST.to_be_bytes().to_vec(), @@ -434,10 +458,10 @@ impl Shared { let hello = { let mut state = self.state.lock().unwrap(); state.members.remove(&slot); - state - .streams - .remove(&slot) - .and_then(|s| s.peer.map(|p| p.hello)) + state.streams.remove(&slot).and_then(|s| { + *s.line.sealer.lock().unwrap() = None; + s.peer.map(|p| p.hello) + }) }; if let Some(hello) = hello { (self.events)(Event::Left(&hello)); diff --git a/tools/TESTING.md b/tools/TESTING.md index e5c19b193d5e6930e2ca7da68b3798e17bef654e..ede71e01e1ca73c17b5be5b0d5372dad95479142 100644 --- a/tools/TESTING.md +++ b/tools/TESTING.md @@ -20,6 +20,7 @@ another, and the exit status is the result. | `windows-aarch64`, `linux-*` | Clippy `-D warnings` on what ships (libraries and binaries but `mobile`), then the `snowbound` build; Linux through `platform/linux/cargo.sh`, which links with zig against glibc 2.17 | | `ios` | `xcodebuild` of the simulator app, unsigned | | `web` | Clippy `-D warnings` on `snowbound` for `wasm32-unknown-unknown`, SQLite built by nixpkgs' clang; `release_web.py` links, optimizes and deploys the static folder | +| `web-js` | Node boundary tests for Live Share browser sockets | | `macos-10.6` | `platform/snow-leopard/cargo.sh` build of `snowbound`; skipped, saying why, without the SDK or nightly `rust-src` | ```sh diff --git a/tools/ci.py b/tools/ci.py index 60cfbb1f633ef25114e8348277ca7a79ca830e27..7cc1135851170accbd8fcf3a9fdf2ba2a5c55219 100644 --- a/tools/ci.py +++ b/tools/ci.py @@ -78,6 +78,10 @@ def lanes(): commands=[cross(linux, arch, *cross_clippy), cross(linux, arch, 'build', '-p', 'snowbound')], environment={'CARGO_TARGET_DIR': str(TARGET / 'linux')}, missing=None if shutil.which('zig') else 'needs zig')) + result.append(dict( + name='web-js', minutes=5, packages=['notebook'], paths=('tools/web/',), + commands=[['node', '--test', 'tools/web/test_live.mjs']], + missing=None if shutil.which('node') else 'needs node')) targets = subprocess.run(['rustup', 'target', 'list', '--installed'], capture_output=True, text=True).stdout wasm = web_environment() result.append(dict( diff --git a/tools/web/test_live.mjs b/tools/web/test_live.mjs new file mode 100644 index 0000000000000000000000000000000000000000..b36012b7dfeb59ccc29ba6c15f9a1ecefa5c5a37 --- /dev/null +++ b/tools/web/test_live.mjs @@ -0,0 +1,90 @@ +import assert from 'node:assert/strict'; +import { readFile } from 'node:fs/promises'; +import test from 'node:test'; + +const opened = []; +class WebSocket { + static OPEN = 1; + constructor() { + this.readyState = WebSocket.OPEN; + this.bufferedAmount = 0; + this.sent = []; + this.closes = 0; + opened.push(this); + } + send(bytes) { + if (this.failure) throw this.failure; + this.sent.push(bytes.slice()); + this.bufferedAmount += bytes.length; + } + close() { + this.closes++; + this.readyState = 3; + } +} +globalThis.WebSocket = WebSocket; +const source = await readFile(new URL('../../crates/notebook/src/live/web.js', import.meta.url), 'utf8'); +const { liveConnect, liveSend, liveClose } = await import( + `data:text/javascript;base64,${Buffer.from(source).toString('base64')}` +); + +for (const accepted of [0, 1]) { + test(`backpressure after ${accepted} chunks retires the socket before notifying`, () => { + let disconnected = 0; + const id = liveConnect('ws://disposable.invalid', () => {}, () => { + disconnected++; + assert.throws(() => liveSend(id, new Uint8Array(1)), /Relay disconnected/); + }); + const socket = opened.at(-1); + const onclose = socket.onclose; + const chunk = new Uint8Array(64 << 10); + socket.bufferedAmount = accepted ? 1048576 - chunk.length + 1 : 1048577; + if (accepted) liveSend(id, chunk); + assert.throws(() => liveSend(id, chunk), /Relay cannot keep up/); + assert.equal(socket.sent.length, accepted); + assert.equal(socket.closes, 1); + assert.equal(disconnected, 1); + assert.equal(socket.onmessage, null); + assert.equal(socket.onclose, null); + onclose(); + liveClose(id); + assert.equal(socket.closes, 1); + assert.equal(disconnected, 1); + + const replacement = liveConnect('ws://disposable.invalid', () => {}, () => disconnected++); + liveSend(replacement, chunk); + assert.throws(() => liveSend(id, chunk), /Relay disconnected/); + onclose(); + assert.equal(opened.at(-1).sent.length, 1); + assert.equal(opened.at(-1).closes, 0); + assert.equal(disconnected, 1); + liveClose(replacement); + }); +} + +test('a WebSocket send exception retires the socket and preserves the error', () => { + let disconnected = 0; + const id = liveConnect('ws://disposable.invalid', () => {}, () => disconnected++); + const socket = opened.at(-1); + socket.failure = new Error('Transport failed'); + assert.throws(() => liveSend(id, new Uint8Array(1)), error => error === socket.failure); + assert.equal(socket.sent.length, 0); + assert.equal(socket.closes, 1); + assert.equal(disconnected, 1); +}); + +test('a non-open socket is retired, while deliberate close never reconnects', () => { + let disconnected = 0; + const id = liveConnect('ws://disposable.invalid', () => {}, () => disconnected++); + const socket = opened.at(-1); + socket.readyState = 2; + assert.throws(() => liveSend(id, new Uint8Array(1)), /Relay disconnected/); + assert.equal(socket.closes, 1); + assert.equal(disconnected, 1); + const next = liveConnect('ws://disposable.invalid', () => {}, () => disconnected++); + const onclose = opened.at(-1).onclose; + liveClose(next); + onclose(); + assert.equal(opened.at(-1).closes, 1); + assert.equal(disconnected, 1); +}); -- 2.54.0