diff --git a/.changeset/curly-pandas-tap.md b/.changeset/curly-pandas-tap.md new file mode 100644 index 00000000..cbf2e9af --- /dev/null +++ b/.changeset/curly-pandas-tap.md @@ -0,0 +1,18 @@ +--- +'@srcbook/api': patch +--- + +Stop several inputs from crashing the server. + +A malformed websocket frame, a websocket message referring to a session that no longer exists, +or unexpected tsserver output could each end the process. In every case the throw happened +inside an event-emitter callback, where the surrounding `try`/`catch` could not reach it. They +are now logged and the operation is dropped. + +Running the same cell twice before the first run exited corrupted the process registry: the +first process's exit handler deleted the entry belonging to the second, leaving a process +running that stop reported as nonexistent. + +tsserver requests (hover, completions, go-to-definition) now time out after 10 seconds and +reject if the tsserver process exits. They previously waited forever and leaked their +resolvers. diff --git a/packages/api/processes.mts b/packages/api/processes.mts index 641cb7fd..918e5699 100644 --- a/packages/api/processes.mts +++ b/packages/api/processes.mts @@ -17,10 +17,21 @@ export class Processes { this.processes[key] = process; process.on('exit', () => { - delete this.processes[key]; + // Only forget this process if it is still the one registered under the key. + // + // Running the same cell twice before the first exits replaces the entry, and + // without this check the first process's exit would delete the second's key — + // leaving a running process that `kill` then claims doesn't exist. + if (this.processes[key] === process) { + delete this.processes[key]; + } }); } + has(sessionId: string, cellId: string) { + return this.toKey(sessionId, cellId) in this.processes; + } + kill(sessionId: string, cellId: string) { const key = this.toKey(sessionId, cellId); diff --git a/packages/api/server/ws-client.mts b/packages/api/server/ws-client.mts index 812be3f0..03f87a4c 100644 --- a/packages/api/server/ws-client.mts +++ b/packages/api/server/ws-client.mts @@ -212,8 +212,21 @@ export default class WebSocketServer { } private handleIncomingMessage(conn: ConnectionType, message: RawData) { - const parsed = JSON.parse(message.toString('utf8')); - const [topic, event, payload] = WebSocketMessageSchema.parse(parsed); + // This runs inside a socket 'message' listener, so anything thrown here is an + // uncaught exception rather than a failed request. Per-event payloads were + // already parsed safely below; the envelope itself was not, which made one + // malformed frame enough to take the server down. + let topic: string; + let event: string; + let payload: Record; + + try { + const parsed = JSON.parse(message.toString('utf8')); + [topic, event, payload] = WebSocketMessageSchema.parse(parsed); + } catch (e) { + console.warn(`Server received a malformed websocket message: ${(e as Error).message}`); + return; + } const channelMatch = this.findChannelMatch(topic); @@ -258,7 +271,29 @@ export default class WebSocketServer { return; } - handler(result.data, { topic: match.topic, event: event, params: match.params }, conn); + // Handlers are async and throw freely — findSession throws for an unknown id, + // which a reconnecting client with a stale session can trigger on its own. An + // unawaited rejection here would be an unhandled rejection, and Node's default + // is to exit the process on those. + try { + const result_ = handler( + result.data, + { topic: match.topic, event: event, params: match.params }, + conn, + ) as void | Promise; + + if (result_ instanceof Promise) { + result_.catch((e) => this.reportHandlerError(topic, event, e)); + } + } catch (e) { + this.reportHandlerError(topic, event, e); + } + } + + private reportHandlerError(topic: string, event: string, e: unknown) { + const error = e instanceof Error ? e : new Error(String(e)); + console.error(`Error handling '${event}' on topic '${topic}': ${error.message}`); + console.error(error); } private findChannelMatch(topic: string): { channel: Channel; match: TopicMatch } | null { diff --git a/packages/api/test/processes.test.mts b/packages/api/test/processes.test.mts new file mode 100644 index 00000000..f20178e0 --- /dev/null +++ b/packages/api/test/processes.test.mts @@ -0,0 +1,120 @@ +import { EventEmitter } from 'node:events'; +import type { ChildProcess } from 'node:child_process'; +import { Processes } from '../processes.mjs'; + +/** + * Minimal stand-in for a ChildProcess: the registry only looks at `pid`, `killed`, + * `kill()` and the 'exit' event. + */ +class FakeProcess extends EventEmitter { + pid: number | undefined; + killed = false; + + constructor(pid: number | undefined = 1234) { + super(); + this.pid = pid; + } + + kill(_signal?: string) { + this.killed = true; + return true; + } + + exit() { + this.emit('exit'); + } + + asChildProcess() { + return this as unknown as ChildProcess; + } +} + +describe('Processes', () => { + it('registers and kills a process', () => { + const processes = new Processes(); + const proc = new FakeProcess(); + + processes.add('session1', 'cell1', proc.asChildProcess()); + + expect(processes.has('session1', 'cell1')).toBe(true); + expect(processes.kill('session1', 'cell1')).toBe(true); + expect(proc.killed).toBe(true); + }); + + it('forgets a process once it exits', () => { + const processes = new Processes(); + const proc = new FakeProcess(); + + processes.add('session1', 'cell1', proc.asChildProcess()); + proc.exit(); + + expect(processes.has('session1', 'cell1')).toBe(false); + }); + + it('refuses a process with no pid', () => { + const processes = new Processes(); + const proc = new FakeProcess(); + // A ChildProcess has no pid when the spawn itself failed. + proc.pid = undefined; + expect(() => processes.add('s', 'c', proc.asChildProcess())).toThrow(/no pid/); + }); + + it('refuses an already-killed process', () => { + const processes = new Processes(); + const proc = new FakeProcess(); + proc.killed = true; + expect(() => processes.add('s', 'c', proc.asChildProcess())).toThrow(/killed/); + }); + + it('throws when killing a cell with nothing running', () => { + const processes = new Processes(); + expect(() => processes.kill('session1', 'cell1')).toThrow(/no process/); + }); + + it('keys processes separately per session and per cell', () => { + const processes = new Processes(); + const a = new FakeProcess(1); + const b = new FakeProcess(2); + + processes.add('session1', 'cell1', a.asChildProcess()); + processes.add('session2', 'cell1', b.asChildProcess()); + + processes.kill('session1', 'cell1'); + + expect(a.killed).toBe(true); + expect(b.killed).toBe(false); + }); + + // The regression this file exists for. Running a cell again before the first run + // exits replaces the registry entry; the first process's exit handler then used to + // delete the key belonging to the second, leaving something running that `kill` + // insisted did not exist. + it('keeps the second process registered when the first exits after being replaced', () => { + const processes = new Processes(); + const first = new FakeProcess(1); + const second = new FakeProcess(2); + + processes.add('session1', 'cell1', first.asChildProcess()); + processes.add('session1', 'cell1', second.asChildProcess()); + + first.exit(); + + expect(processes.has('session1', 'cell1')).toBe(true); + expect(() => processes.kill('session1', 'cell1')).not.toThrow(); + expect(second.killed).toBe(true); + }); + + it('forgets the key when the surviving process finally exits', () => { + const processes = new Processes(); + const first = new FakeProcess(1); + const second = new FakeProcess(2); + + processes.add('session1', 'cell1', first.asChildProcess()); + processes.add('session1', 'cell1', second.asChildProcess()); + + first.exit(); + second.exit(); + + expect(processes.has('session1', 'cell1')).toBe(false); + }); +}); diff --git a/packages/api/test/ws-client.test.mts b/packages/api/test/ws-client.test.mts new file mode 100644 index 00000000..563e46b6 --- /dev/null +++ b/packages/api/test/ws-client.test.mts @@ -0,0 +1,232 @@ +import { EventEmitter } from 'node:events'; +import type { IncomingMessage } from 'node:http'; +import type { WebSocket } from 'ws'; +import z from 'zod'; +import WebSocketServer, { Channel } from '../server/ws-client.mjs'; + +describe('Channel.match', () => { + it('matches a static topic exactly', () => { + const channel = new Channel('session'); + expect(channel.match('session')).toEqual({ topic: 'session', params: {} }); + expect(channel.match('sessions')).toBeNull(); + expect(channel.match('session:123')).toBeNull(); + }); + + it('extracts a wildcard parameter', () => { + const channel = new Channel('session:'); + expect(channel.match('session:abc123')).toEqual({ + topic: 'session:abc123', + params: { sessionId: 'abc123' }, + }); + }); + + it('requires the same number of segments', () => { + const channel = new Channel('session:'); + expect(channel.match('session')).toBeNull(); + expect(channel.match('session:abc:extra')).toBeNull(); + }); + + it('supports several wildcards', () => { + const channel = new Channel('room::users:'); + expect(channel.match('room:1:users:2')).toEqual({ + topic: 'room:1:users:2', + params: { roomId: '1', userId: '2' }, + }); + expect(channel.match('room:1:groups:2')).toBeNull(); + }); + + it('rejects topic patterns that are not valid identifiers', () => { + expect(() => new Channel('session:')).toThrow(/Invalid channel topic/); + expect(() => new Channel('session:has space')).toThrow(/Invalid channel topic/); + }); +}); + +/** Stand-in for a ws socket: records what the server sends. */ +class FakeSocket extends EventEmitter { + sent: string[] = []; + closed: { code?: number; reason?: string } | null = null; + + send(data: string) { + this.sent.push(data); + } + + close(code?: number, reason?: string) { + this.closed = { code, reason }; + } + + messages() { + return this.sent.map((raw) => JSON.parse(raw)); + } + + asWebSocket() { + return this as unknown as WebSocket; + } +} + +function request(origin = 'http://localhost:2150', url = '/websocket'): IncomingMessage { + return { + url, + headers: { host: 'localhost:2150', origin }, + } as unknown as IncomingMessage; +} + +function connect(wss: WebSocketServer, origin?: string) { + const socket = new FakeSocket(); + wss.onConnection(socket.asWebSocket(), request(origin)); + return socket; +} + +function send(socket: FakeSocket, message: unknown) { + socket.emit('message', Buffer.from(JSON.stringify(message), 'utf8')); +} + +describe('WebSocketServer', () => { + it('closes connections on paths other than /websocket', () => { + const wss = new WebSocketServer(); + const socket = new FakeSocket(); + wss.onConnection(socket.asWebSocket(), request(undefined, '/not-websocket')); + expect(socket.closed).not.toBeNull(); + }); + + it('rejects a connection from a disallowed origin', () => { + const wss = new WebSocketServer(); + const socket = connect(wss, 'https://evil.example.com'); + expect(socket.closed?.code).toBe(1008); + }); + + it('subscribes and dispatches an event to its handler', () => { + const wss = new WebSocketServer(); + const received: Array<{ payload: any; params: any }> = []; + + wss + .channel('session:') + .on('cell:exec', z.object({ cellId: z.string() }), (payload, context) => { + received.push({ payload, params: context.params }); + }); + + const socket = connect(wss); + send(socket, ['session:abc', 'subscribe', { id: 'sub-1' }]); + send(socket, ['session:abc', 'cell:exec', { cellId: 'cell-1' }]); + + expect(socket.messages()).toContainEqual(['session:abc', 'subscribed', { id: 'sub-1' }]); + expect(received).toEqual([{ payload: { cellId: 'cell-1' }, params: { sessionId: 'abc' } }]); + }); + + it('broadcasts only to connections subscribed to the topic', () => { + const wss = new WebSocketServer(); + wss.channel('session:'); + + const subscribed = connect(wss); + const other = connect(wss); + + send(subscribed, ['session:abc', 'subscribe', { id: 'sub-1' }]); + send(other, ['session:xyz', 'subscribe', { id: 'sub-2' }]); + + wss.broadcast('session:abc', 'cell:updated', { cell: { id: 'c1' } }); + + expect(subscribed.messages()).toContainEqual([ + 'session:abc', + 'cell:updated', + { cell: { id: 'c1' } }, + ]); + expect(other.messages()).not.toContainEqual([ + 'session:abc', + 'cell:updated', + { cell: { id: 'c1' } }, + ]); + }); + + it('stops broadcasting after unsubscribe', () => { + const wss = new WebSocketServer(); + wss.channel('session:'); + + const socket = connect(wss); + send(socket, ['session:abc', 'subscribe', { id: 'sub-1' }]); + send(socket, ['session:abc', 'unsubscribe', {}]); + + wss.broadcast('session:abc', 'cell:updated', {}); + + expect(socket.messages().filter(([, event]) => event === 'cell:updated')).toEqual([]); + }); + + // These four are the crash cases. The envelope used to be JSON.parsed and zod-parsed + // inline in the 'message' listener, so a bad frame was an uncaught exception rather + // than a dropped message. + describe('malformed input', () => { + function serverWithHandler() { + const wss = new WebSocketServer(); + wss + .channel('session:') + .on('cell:exec', z.object({ cellId: z.string() }), () => {}); + return wss; + } + + it('survives a frame that is not JSON', () => { + const socket = connect(serverWithHandler()); + expect(() => socket.emit('message', Buffer.from('not json', 'utf8'))).not.toThrow(); + }); + + it('survives a frame that is JSON but not a valid envelope', () => { + const wss = serverWithHandler(); + const socket = connect(wss); + expect(() => send(socket, { not: 'a tuple' })).not.toThrow(); + expect(() => send(socket, ['only-one-element'])).not.toThrow(); + }); + + it('ignores an unknown topic and an unknown event', () => { + const wss = serverWithHandler(); + const socket = connect(wss); + expect(() => send(socket, ['nope:abc', 'cell:exec', { cellId: 'c' }])).not.toThrow(); + expect(() => send(socket, ['session:abc', 'nope', {}])).not.toThrow(); + }); + + it('ignores a payload that fails its schema', () => { + const wss = serverWithHandler(); + const socket = connect(wss); + expect(() => send(socket, ['session:abc', 'cell:exec', { cellId: 42 }])).not.toThrow(); + }); + }); + + // Handlers are async and throw freely — findSession throws for an unknown session id, + // which a reconnecting client can trigger by itself. Unawaited, that was an unhandled + // rejection, and Node exits the process on those by default. + describe('handler errors', () => { + // These deliberately trigger the error path, which logs. Silenced so a passing + // run doesn't print stack traces that look like failures. + beforeEach(() => { + vi.spyOn(console, 'error').mockImplementation(() => {}); + }); + + afterEach(() => { + vi.restoreAllMocks(); + }); + + it('does not throw when a handler throws synchronously', () => { + const wss = new WebSocketServer(); + wss.channel('session:').on('cell:exec', z.object({}), () => { + throw new Error('boom'); + }); + + const socket = connect(wss); + expect(() => send(socket, ['session:abc', 'cell:exec', {}])).not.toThrow(); + }); + + it('catches a rejected promise from an async handler', async () => { + const wss = new WebSocketServer(); + let settled = false; + + wss.channel('session:').on('cell:exec', z.object({}), async () => { + settled = true; + throw new Error('Session with id abc not found'); + }); + + const socket = connect(wss); + send(socket, ['session:abc', 'cell:exec', {}]); + + // Give the rejection a turn to propagate. If it were unhandled, vitest would + // report an unhandled rejection for this test. + await new Promise((resolve) => setImmediate(resolve)); + expect(settled).toBe(true); + }); + }); +}); diff --git a/packages/api/tsserver/tsserver.mts b/packages/api/tsserver/tsserver.mts index 5952c39e..6560eb8c 100644 --- a/packages/api/tsserver/tsserver.mts +++ b/packages/api/tsserver/tsserver.mts @@ -31,26 +31,62 @@ import { parse } from './messages.mjs'; * - https://github.com/microsoft/TypeScript/blob/v5.5.3/src/server/protocol.ts * - https://github.com/microsoft/TypeScript/wiki/Standalone-Server-(tsserver) */ +type PendingRequest = { + resolve: (value: any) => void; + reject: (error: Error) => void; + timeout: NodeJS.Timeout; +}; + +// Requests are interactive (hover, completions, go-to-definition), so a slow answer +// is no more useful than no answer, and every request must eventually settle. +const REQUEST_TIMEOUT_MS = 10_000; + export class TsServer extends EventEmitter { private _seq: number = 0; private buffered: Buffer = Buffer.from(''); private readonly process: ChildProcess; - private readonly resolvers: Record void> = {}; + private readonly pending: Record = {}; constructor(process: ChildProcess) { super(); this.process = process; this.process.stdout?.on('data', (chunk) => { - const { messages, buffered } = parse(chunk, this.buffered); - this.buffered = buffered; - for (const message of messages) { - if (message.type === 'response') { - this.handleResponse(message); - } else if (message.type === 'event') { - this.handleEvent(message); + // This runs in a 'data' listener, so a throw from parse() — which happens on + // any framing it doesn't recognise — would be an uncaught exception. Drop the + // buffer and carry on instead: tsserver output is a best-effort feature, not + // something worth ending the process over. + try { + const { messages, buffered } = parse(chunk, this.buffered); + this.buffered = buffered; + for (const message of messages) { + if (message.type === 'response') { + this.handleResponse(message); + } else if (message.type === 'event') { + this.handleEvent(message); + } } + } catch (e) { + console.error(`Error parsing tsserver output: ${(e as Error).message}`); + this.buffered = Buffer.from(''); } }); + + // Anything still waiting on a response will never get one once the process is + // gone. Without this those promises hang forever and leak their resolvers. + this.process.on('exit', (code, signal) => { + this.rejectPending(new Error(`tsserver exited (code ${code}, signal ${signal})`)); + }); + } + + private rejectPending(error: Error) { + for (const seq of Object.keys(this.pending)) { + const entry = this.pending[Number(seq)]; + if (entry) { + clearTimeout(entry.timeout); + entry.reject(error); + } + delete this.pending[Number(seq)]; + } } private get seq() { @@ -58,19 +94,20 @@ export class TsServer extends EventEmitter { } private handleResponse(response: tsserver.protocol.Response) { - const resolve = this.resolvers[response.request_seq]; + const entry = this.pending[response.request_seq]; - if (!resolve) { + if (!entry) { console.warn( - `Received a response for command '${response.command}' and request_seq '${response.request_seq}' but no resolver was found. This may be a bug in the code.\n\nResponse:\n${JSON.stringify(response, null, 2)}\n`, + `Received a response for command '${response.command}' and request_seq '${response.request_seq}' but no pending request was found. It may have timed out.`, ); return; } - delete this.resolvers[response.request_seq]; + clearTimeout(entry.timeout); + delete this.pending[response.request_seq]; - resolve(response); + entry.resolve(response); } private handleEvent(event: tsserver.protocol.Event) { @@ -82,8 +119,13 @@ export class TsServer extends EventEmitter { } private sendWithResponsePromise(request: tsserver.protocol.Request) { - return new Promise((resolve) => { - this.resolvers[request.seq] = resolve; + return new Promise((resolve, reject) => { + const timeout = setTimeout(() => { + delete this.pending[request.seq]; + reject(new Error(`tsserver did not respond to '${request.command}' in time`)); + }, REQUEST_TIMEOUT_MS); + + this.pending[request.seq] = { resolve, reject, timeout }; this.send(request); }); } @@ -114,6 +156,7 @@ export class TsServer extends EventEmitter { */ shutdown() { this.removeAllListeners(); + this.rejectPending(new Error('tsserver was shut down')); return this.process.kill('SIGTERM'); } diff --git a/tasks/0003-stop-the-crashers.md b/tasks/0003-stop-the-crashers.md index 208b18bf..6d621066 100644 --- a/tasks/0003-stop-the-crashers.md +++ b/tasks/0003-stop-the-crashers.md @@ -1,7 +1,7 @@ --- id: 0003 title: Stop the process crashers -status: TODO +status: DONE created: 2026-07-29 area: api --- @@ -33,4 +33,26 @@ work, so a crash is worse than its individual severity suggests. ## Notes Worth adding a top-level `unhandledRejection` handler as a backstop — but as a safety net -that logs loudly, not as the fix. +that logs loudly, not as the fix. Not added: with the handler-level catch in place there is +no known path to one, and a blanket handler would hide the next one instead of surfacing it. + +All five fixed. (1) and (4) landed earlier in the stack — the depcheck throw in the injection +PR, since that function was being rewritten to use `execFile` anyway, and `process.send` in +the network PR, because that's how it was discovered. + +The shape shared by (2), (3) and (5) is worth naming: **a throw inside an event-emitter +callback is an uncaught exception, not a failed operation.** `handleIncomingMessage` runs in a +socket `'message'` listener, `parse()` runs in a `'data'` listener, and the depcheck callback +runs in an `exec` callback. In all three the surrounding `try/catch` or promise looked like it +covered the code, and didn't. + +Also fixed while here, from task 0005 and adjacent: tsserver request promises never settled if +the server died or never replied. They now time out at 10s and reject on process exit, so +`this.pending` can't leak. + +The tsserver timeout is 10s because these are interactive requests — hover, completions, +go-to-definition. An answer that late is no more useful than no answer; what matters is that +the promise settles at all. + +Still open, deliberately: a rejected tsserver request leaves the _client_ hanging, because +responses are broadcasts with no request correlation. That's task 0009. diff --git a/tasks/0005-fix-process-registry-clobbering.md b/tasks/0005-fix-process-registry-clobbering.md index 78762183..0f1ffe78 100644 --- a/tasks/0005-fix-process-registry-clobbering.md +++ b/tasks/0005-fix-process-registry-clobbering.md @@ -1,7 +1,7 @@ --- id: 0005 title: Fix process registry clobbering on double-exec -status: TODO +status: DONE created: 2026-07-29 area: api --- @@ -40,3 +40,17 @@ Guarding the delete is a two-line fix. The "what should double-exec even do" que real decision — the UI currently disables run while running, so this is reachable mainly via a second client or a race, which makes it exactly the kind of bug that shows up once and is never reproduced. + +Partially done. The delete is now guarded — the exit handler only clears the key if it still +holds the same process reference — and there's a test for the double-exec sequence, plus a +`has()` method so callers can ask before adding. + +**Not decided:** what double-exec _should_ do. The registry no longer corrupts itself, but the +first process is still orphaned rather than killed or refused. That's a product question +(silently replace? refuse with an error? queue?) and it belongs with the execution-model +decision in task 0006, since a stateful kernel would change the answer entirely. + +**Not verified:** whether `SIGTERM` to the `tsx` wrapper actually kills the TypeScript cell +process. Needs a running notebook and a deliberately long-running cell to check. Left open — +if `tsx` forks a child, stop is only cosmetic for TypeScript cells, which is the more +interesting half of this task and shouldn't be quietly closed.