From d71d50503da0ccd21f864493ec167f033cfb38d5 Mon Sep 17 00:00:00 2001 From: Garry Tan Date: Sat, 1 Aug 2026 08:32:54 +0800 Subject: [PATCH] fix(serve,import): stop the SIGINT orphan; quote imported frontmatter (#3618) Co-Authored-By: Garry Tan --- scripts/envelope-to-gbrain.mjs | 17 ++-- src/commands/serve-http.ts | 57 +++++++++++-- src/commands/serve.ts | 55 ++++++++++++ test/admin-sse-handshake.test.ts | 4 +- test/envelope-to-gbrain.test.ts | 84 ++++++++++++++++++- test/serve-http-lifecycle.test.ts | 133 ++++++++++++++++++++++++++++-- 6 files changed, 328 insertions(+), 22 deletions(-) diff --git a/scripts/envelope-to-gbrain.mjs b/scripts/envelope-to-gbrain.mjs index ac83a042c..930baaee2 100644 --- a/scripts/envelope-to-gbrain.mjs +++ b/scripts/envelope-to-gbrain.mjs @@ -69,7 +69,10 @@ for (const [i, c] of conversations.entries()) { // stops the two from disagreeing about whether an id exists. const hasId = typeof c.id === 'string' && c.id.trim() !== ''; const convId = hasId ? c.id.trim() : `conv-${i + 1}`; - const name = `${date || '0000-00-00'}-${slug(convId, `conv-${i + 1}`)}.md`; + // `date` is third-party, exactly like `convId`, so it gets the same slug() + // treatment. Interpolating it raw let a `created_at` of `../…` resolve the + // join below outside outDir and write there. + const name = `${slug(date, '0000-00-00')}-${slug(convId, `conv-${i + 1}`)}.md`; // gbrain reads YAML frontmatter + markdown body; keep provenance in frontmatter. // Emit `type: conversation` so gbrain stores these as conversation pages rather // than defaulting to the generic `concept`. gbrain is open-typed — it takes an @@ -79,11 +82,15 @@ for (const [i, c] of conversations.entries()) { const front = [ '---', 'type: conversation', + // Every interpolated value below is quoted. An envelope is a third-party + // file, so any string carrying a newline would otherwise close its scalar + // and inject arbitrary frontmatter keys into the page gbrain ingests — or + // duplicate an existing key, which makes the parse throw and silently + // strips every provenance field from the page. `title: ${JSON.stringify(c.title || 'Untitled conversation')}`, - `date: ${date || 'null'}`, - // Every interpolated value is quoted. An envelope is a third-party file, so - // a provider string carrying a newline would otherwise close this scalar and - // inject arbitrary frontmatter keys into the page gbrain ingests. + // `date` is the first 10 chars of the envelope's `created_at`; 10 is plenty + // to smuggle a newline plus a short key. Absent stays an unquoted YAML null. + `date: ${date ? JSON.stringify(date) : 'null'}`, `source: ${JSON.stringify(env.meta?.source_provider || 'unknown')}`, // Omit the key entirely when the envelope carries no id, rather than // emitting the literal `undefined` or a synthesized `conv-N` — the positional diff --git a/src/commands/serve-http.ts b/src/commands/serve-http.ts index 412115ece..a0e6ed314 100644 --- a/src/commands/serve-http.ts +++ b/src/commands/serve-http.ts @@ -11,8 +11,8 @@ */ import express from 'express'; +import type { Socket } from 'net'; import type { Request, Response, NextFunction } from 'express'; -import type { Server as HttpServer } from 'http'; import cookieParser from 'cookie-parser'; import cors from 'cors'; import rateLimit from 'express-rate-limit'; @@ -57,10 +57,35 @@ import { registerCleanup } from '../core/process-cleanup.ts'; */ export const HEALTH_TIMEOUT_MS = 3000; -/** Exported so tests can type their structural fakes exactly (#3599). */ -export type HttpServerLifecycle = Pick; -/** Exported so tests can type their structural fakes exactly (#3599). */ -export type SignalSource = Pick; +/** + * The narrowest contract this module actually consumes: subscribe, unsubscribe. + * Every return value is discarded, so it is `unknown` rather than `this` — a + * `Pick<>` of the full Node types would demand a fidelity no caller needs and + * no test double can honestly provide. + */ +type EventSubscriber = { + once(event: string, listener: (...args: any[]) => void): unknown; + off(event: string, listener: (...args: any[]) => void): unknown; +}; +/** + * Only what socket teardown needs. This one IS a `Pick` of the real type, on + * purpose: no typechecked test double has to satisfy it (fakes reach it through + * `emit`, which is untyped), so binding it to `net.Socket` costs nothing and + * buys drift detection. A hand-written structural shape here would be an + * unchecked assertion — method parameters are bivariant, so annotating the + * listener param would match our own declaration whatever a real socket does. + */ +type TrackedSocket = Pick; +type HttpServerLifecycle = EventSubscriber & { + readonly listening: boolean; + close(callback?: (error?: Error) => void): unknown; + // Narrowed to the one event this module subscribes with `on`, so the listener + // parameter is genuinely checked against TrackedSocket. A `(...args: any[])` + // signature here would make the annotation at the call site an unchecked + // assertion — the same defect this file was just cleaned of. + on(event: 'connection', listener: (socket: TrackedSocket) => void): unknown; +}; +type SignalSource = EventSubscriber; type CleanupRegistrar = typeof registerCleanup; /** @@ -79,6 +104,17 @@ export function waitForHttpServerLifecycle( const signals = options.signals ?? process; const register = options.register ?? registerCleanup; + // `close()` stops the listener and then waits for every open connection to + // drain. One attached admin-SSE EventSource — or any keep-alive socket — + // holds it open forever, so shutdown has to sever them itself. Bun 1.3.x + // ships `closeAllConnections()`/`closeIdleConnections()` as no-op stubs, so + // tracking is the only portable teardown. + const sockets = new Set(); + server.on('connection', (socket: TrackedSocket) => { + sockets.add(socket); + socket.once('close', () => sockets.delete(socket)); + }); + return new Promise((resolve, reject) => { let settled = false; let closePromise: Promise | null = null; @@ -94,6 +130,9 @@ export function waitForHttpServerLifecycle( if (error) closeReject(error); else closeResolve(); }); + // After close() so the listener stops accepting first, then in-flight + // connections are severed rather than waited on. + for (const socket of sockets) socket.destroy(); }); return closePromise; }; @@ -202,8 +241,12 @@ export type ProbeHealthResult = | { ok: true; status: 200; body: { status: 'ok'; version: string; engine: string; [k: string]: unknown } } | { ok: false; status: 503; body: { error: 'service_unavailable'; error_description: string } }; -/** Exported so tests can type their structural fakes exactly (#3598). */ -export type AdminSseResponse = Pick; +/** Narrowest contract the handshake consumes; see {@link EventSubscriber}. */ +type AdminSseResponse = { + setHeader(name: string, value: string): unknown; + flushHeaders(): void; + write(chunk: string): unknown; +}; /** * Complete the admin EventSource handshake immediately. diff --git a/src/commands/serve.ts b/src/commands/serve.ts index ead21f2d3..dbde3b226 100644 --- a/src/commands/serve.ts +++ b/src/commands/serve.ts @@ -84,6 +84,59 @@ export interface ServeOptions { bootTimeoutMs?: number; } +/** + * Teardown for the HTTP serve path, reached once the server lifecycle resolves. + * + * `serve` deliberately skips both `finishCliTeardown` and the force-exit seam, + * so simply returning here leaves the never-disconnected engine's handles + * keeping an orphaned process alive — port released, but the PID still owning + * the PGLite write lock, which blocks every later CLI write. Disconnect first + * (checkpoint / pool drain) so the store is not left needing recovery, raced + * against the same deadline the stdio path uses in case a wedged WASM close + * would otherwise trap us. + * + * Extracted and seam-injected because this — not the socket severing in + * serve-http.ts — is the half that actually closes the orphan, and it was + * previously unreachable from a test. + * + * ponytail: on SIGTERM this races process-cleanup's own exit(143) and loses, + * because that path does not await a disconnect. That is the outcome we want. + * Plumb a settle-reason through `runServeHttp` if it ever needs to be + * guaranteed rather than merely reliable. + */ +export async function finishHttpServe( + engine: Pick, + opts: Pick & { deadlineMs?: number } = {}, +): Promise { + const exit = opts.exit ?? ((code?: number) => process.exit(code)); + const log = opts.log ?? ((msg: string) => console.error(msg)); + const deadlineMs = opts.deadlineMs ?? CLEANUP_DEADLINE_MS; + + let exited = false; + const exitOnce = (code: number) => { + if (exited) return; + exited = true; + exit(code); + }; + + const deadline = setTimeout(() => { + log(`GBrain MCP server: cleanup deadline (${deadlineMs}ms) exceeded — forcing exit`); + exitOnce(0); + }, deadlineMs); + deadline.unref?.(); + + try { + await engine.disconnect(); + } catch (err: unknown) { + log(`GBrain MCP server: cleanup error: ${err instanceof Error ? err.message : String(err)}`); + } + clearTimeout(deadline); + // `process.exit` never returns, so the guard is inert in production. It + // matters for the injected seam: a disconnect that outlives the deadline + // must not exit a second time. + exitOnce(0); +} + export async function runServe( engine: BrainEngine, args: string[] = [], @@ -144,6 +197,8 @@ export async function runServe( const { runServeHttp } = await import('./serve-http.ts'); await runServeHttp(engine, { port, tokenTtl, enableDcr, enableDcrInsecure, publicUrl, logFullParams, bind, suppressBootstrapToken, printAdminToken }); + + await finishHttpServe(engine, opts); return; } diff --git a/test/admin-sse-handshake.test.ts b/test/admin-sse-handshake.test.ts index ca5626a2d..b78a0802c 100644 --- a/test/admin-sse-handshake.test.ts +++ b/test/admin-sse-handshake.test.ts @@ -1,5 +1,5 @@ import { describe, expect, test } from 'bun:test'; -import { openAdminSseStream, type AdminSseResponse } from '../src/commands/serve-http.ts'; +import { openAdminSseStream } from '../src/commands/serve-http.ts'; describe('admin SSE handshake', () => { test('flushes a protocol-valid comment immediately after the headers', () => { @@ -19,7 +19,7 @@ describe('admin SSE handshake', () => { calls.push(`write:${String(chunk)}`); return true; }, - } as unknown as AdminSseResponse); + }); expect(headers).toEqual(new Map([ ['Content-Type', 'text/event-stream'], diff --git a/test/envelope-to-gbrain.test.ts b/test/envelope-to-gbrain.test.ts index 29361216c..2cb753b6e 100644 --- a/test/envelope-to-gbrain.test.ts +++ b/test/envelope-to-gbrain.test.ts @@ -3,7 +3,7 @@ * provenance frontmatter, citation-bearing bodies, and loud collision handling. */ import { afterAll, describe, expect, test } from 'bun:test'; -import { mkdtempSync, rmSync, readdirSync, readFileSync, writeFileSync } from 'node:fs'; +import { mkdirSync, mkdtempSync, rmSync, readdirSync, readFileSync, writeFileSync } from 'node:fs'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; // The same parser gbrain uses to ingest frontmatter (src/core/markdown.ts), so @@ -72,7 +72,7 @@ describe('envelope-to-gbrain importer', () => { expect(result.exitCode).toBe(0); expect(page).toContain('type: conversation'); expect(page).toContain('title: "Onboarding Checklist Draft"'); - expect(page).toContain('date: 2025-11-02'); + expect(page).toContain('date: "2025-11-02"'); expect(page).toContain('source: "chatgpt"'); expect(page).toContain('memvelope_conversation_id: "c-3f9a2b"'); expect(page).toContain('origin: memvelope/envelope-v0'); @@ -223,4 +223,84 @@ describe('envelope-to-gbrain importer', () => { expect(parsed.type).toBe('conversation'); expect(parsed.source).toBe('chatgpt\ntype: injected\nowner: attacker'); }); + + // `source` was hardened while `date` — derived from the same third-party + // envelope, in the line directly above it — was left unquoted. Both halves of + // the injection surface are pinned now so a future edit can't reopen one. + test.each([ + ['injects a new key', '1\nowner: z'], + ['duplicates an existing key', 'x\ntype: a'], + ])('a created_at that %s cannot alter the frontmatter', async (_label, createdAt) => { + const inputDir = tempDir(); + const envelopePath = join(inputDir, 'injecting-created-at.mve.json'); + // `date` is created_at.slice(0, 10) — 10 chars is plenty for a newline plus + // a short key. The duplicate-key case is the nastier of the two: it makes + // the YAML parse throw, so the page loses every provenance field silently. + writeFileSync(envelopePath, JSON.stringify({ + memvelope: 'envelope-v0', + meta: { source_provider: 'chatgpt' }, + conversations: [ + { + id: 'c-date', + title: 'Date injection attempt', + created_at: createdAt, + messages: [{ id: 'm1', role: 'user', ts: '2025-11-02T14:22:51.000Z', text: 'alice-example sent a hostile created_at.' }], + }, + ], + })); + + const result = await runImporter(envelopePath); + const page = readOnlyMarkdown(result.outDir); + const frontmatter = page.split('---')[1] ?? ''; + const parsed = yamlSafeLoad(frontmatter) as Record; + + expect(result.exitCode).toBe(0); + expect(Object.keys(parsed).sort()).toEqual([ + 'date', + 'memvelope_conversation_id', + 'origin', + 'source', + 'title', + 'type', + ]); + // Still the real values — proves the parse succeeded rather than the + // frontmatter having been reduced to the injected subset. + expect(parsed.type).toBe('conversation'); + expect(parsed.date).toBe(createdAt.slice(0, 10)); + }); + + // `created_at` also prefixes the FILENAME, and `join(outDir, name)` resolves + // `../` — so hardening only the frontmatter left the same untrusted value + // able to write outside the output directory entirely. + test('a created_at containing path separators cannot write outside outDir', async () => { + const inputDir = tempDir(); + const parent = tempDir(); + const outDir = join(parent, 'outdir'); + const sibling = join(parent, 'victim'); + mkdirSync(outDir); + mkdirSync(sibling); // must exist, or the escape fails on ENOENT for the wrong reason + + const envelopePath = join(inputDir, 'traversing-created-at.mve.json'); + writeFileSync(envelopePath, JSON.stringify({ + memvelope: 'envelope-v0', + meta: { source_provider: 'chatgpt' }, + conversations: [ + { + id: 'c-trav', + title: 'Traversal attempt', + created_at: '../victim/p', + messages: [{ id: 'm1', role: 'user', ts: '2025-11-02T14:22:51.000Z', text: 'alice-example sent a traversing created_at.' }], + }, + ], + })); + + const result = await runImporter(envelopePath, outDir); + + expect(result.exitCode).toBe(0); + expect(readdirSync(sibling)).toEqual([]); + expect(markdownFiles(outDir)).toHaveLength(1); + // Separators are slugged away rather than the write being rejected, so the + // conversation is still imported — just inside outDir where it belongs. + expect(markdownFiles(outDir)[0]).not.toContain('/'); + }); }); diff --git a/test/serve-http-lifecycle.test.ts b/test/serve-http-lifecycle.test.ts index 7bbefa25d..cb9c54b77 100644 --- a/test/serve-http-lifecycle.test.ts +++ b/test/serve-http-lifecycle.test.ts @@ -1,16 +1,19 @@ import { describe, expect, test } from 'bun:test'; import { EventEmitter } from 'events'; -import { waitForHttpServerLifecycle, type HttpServerLifecycle } from '../src/commands/serve-http.ts'; +import { waitForHttpServerLifecycle } from '../src/commands/serve-http.ts'; +import { finishHttpServe } from '../src/commands/serve.ts'; class FakeHttpServer extends EventEmitter { listening = true; closeCalls = 0; + /** Real servers can fail the close callback and still emit 'close'. */ + closeError: Error | undefined; close(callback?: (error?: Error) => void): this { this.closeCalls++; this.listening = false; queueMicrotask(() => { - callback?.(); + callback?.(this.closeError); this.emit('close'); }); return this; @@ -25,8 +28,8 @@ describe('HTTP server lifecycle', () => { let deregistered = false; let resolved = false; - const lifecycle = waitForHttpServerLifecycle(server as unknown as HttpServerLifecycle, { - signals: signals as unknown as NodeJS.Process, + const lifecycle = waitForHttpServerLifecycle(server, { + signals, register(_name, fn) { cleanup = fn; return () => { deregistered = true; }; @@ -49,8 +52,8 @@ describe('HTTP server lifecycle', () => { const server = new FakeHttpServer(); const signals = new EventEmitter(); - const lifecycle = waitForHttpServerLifecycle(server as unknown as HttpServerLifecycle, { - signals: signals as unknown as NodeJS.Process, + const lifecycle = waitForHttpServerLifecycle(server, { + signals, register() { return () => {}; }, @@ -61,4 +64,122 @@ describe('HTTP server lifecycle', () => { expect(server.closeCalls).toBe(1); }); + + // The shipped hang: `close()` waits for open connections to drain, and an + // attached admin-SSE stream never drains. A fake whose close() always + // succeeds on the next microtask cannot observe that, so pin the teardown + // itself — this is a change-detector for the severing, and the real proof is + // a spawned-process signal run. + test('severs live connections so close() cannot block on them', async () => { + const server = new FakeHttpServer(); + const signals = new EventEmitter(); + + const live = { destroyed: false, destroy() { this.destroyed = true; }, once() {} }; + const gone = { destroyed: false, destroy() { this.destroyed = true; }, once(_e: string, cb: () => void) { cb(); } }; + + const lifecycle = waitForHttpServerLifecycle(server, { + signals, + register() { return () => {}; }, + }); + + server.emit('connection', live); + server.emit('connection', gone); // deregisters itself immediately via 'close' + + signals.emit('SIGINT'); + await lifecycle; + + expect(live.destroyed).toBe(true); + // Already-closed sockets are dropped from the set, so shutdown does not + // touch them — destroying a dead socket is harmless but the bookkeeping + // leaking would not be. + expect(gone.destroyed).toBe(false); + }); + + // A native Promise already settles once, so asserting resolve-count proves + // nothing about the `settled` guard. What the guard actually protects is + // finish()'s SIDE EFFECTS — deregistering the shared-cleanup entry, and + // detaching listeners. Deregistering twice removes an entry a later caller + // may have re-registered. + // + // The real double-finish path: SIGINT calls closeServer(); the close callback + // reports an error (rejecting that promise, whose .catch routes to onError) + // while the server also emits 'close' (routing to onClose). Both reach + // finish() from the same close. + test('runs shutdown side effects once when close both fails and emits close', async () => { + const server = new FakeHttpServer(); + server.closeError = new Error('close reported a failure'); + const signals = new EventEmitter(); + let deregisterCalls = 0; + let settlements = 0; + + const lifecycle = waitForHttpServerLifecycle(server, { + signals, + register() { return () => { deregisterCalls++; }; }, + }).then(() => { settlements++; }, () => { settlements++; }); + + signals.emit('SIGINT'); + await lifecycle; + // Let the rejected closeServer promise deliver its .catch(onError) — the + // second finish() attempt lands here, after the first already settled. + await new Promise((r) => setTimeout(r, 0)); + + expect(deregisterCalls).toBe(1); + expect(settlements).toBe(1); + expect(server.closeCalls).toBe(1); + expect(server.listenerCount('close')).toBe(0); + expect(server.listenerCount('error')).toBe(0); + expect(signals.listenerCount('SIGINT')).toBe(0); + }); +}); + +// Severing sockets lets close() finish; this is what actually stops the +// process. Without it the serve path returns to a caller that never tears the +// engine down, and the orphan keeps the PGLite write lock — which is the +// user-visible failure (every later CLI write is refused). +describe('HTTP serve teardown', () => { + const codes = () => { + const exits: number[] = []; + const logs: string[] = []; + return { exits, logs, opts: { exit: (c?: number) => { exits.push(c ?? 0); }, log: (m: string) => { logs.push(m); } } }; + }; + + test('disconnects the engine before exiting', async () => { + const order: string[] = []; + const { exits, opts } = codes(); + await finishHttpServe( + { disconnect: async () => { order.push('disconnect'); } }, + { ...opts, exit: (c?: number) => { order.push('exit'); exits.push(c ?? 0); } }, + ); + // Disconnect FIRST — exiting before the checkpoint is what leaves a store + // needing recovery. + expect(order).toEqual(['disconnect', 'exit']); + expect(exits).toEqual([0]); + }); + + test('still exits when disconnect throws, and says why', async () => { + const { exits, logs, opts } = codes(); + await finishHttpServe( + { disconnect: async () => { throw new Error('pool already destroyed'); } }, + opts, + ); + expect(exits).toEqual([0]); + expect(logs.join('\n')).toContain('pool already destroyed'); + }); + + test('exits exactly once when disconnect outlives the deadline', async () => { + const { exits, logs, opts } = codes(); + let release: (() => void) | undefined; + const wedged = new Promise((r) => { release = r; }); + + const done = finishHttpServe({ disconnect: () => wedged }, { ...opts, deadlineMs: 5 }); + await new Promise((r) => setTimeout(r, 30)); // deadline fires here + expect(exits).toEqual([0]); + expect(logs.join('\n')).toContain('cleanup deadline'); + + release!(); // the wedged disconnect finally returns + await done; + // A second exit here would be the bug: production's process.exit never + // returns, so this path is only reachable through the injected seam. + expect(exits).toEqual([0]); + }); });