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); +});