fix: stop malformed input and lost sessions from crashing the server

.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.

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

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<string, any>;
+
+    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<void>;
+
+      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 {

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