From a14174a726ba54e44cd188609f56e9dd85441e36 Mon Sep 17 00:00:00 2001 From: agent_coder Date: Sat, 11 Jul 2026 02:54:03 +0300 Subject: [PATCH] =?UTF-8?q?fix(ai):=20drain-hang=20=D0=B2=20writeToServerR?= =?UTF-8?q?esponse=20=E2=80=94=20=D1=80=D0=B0=D1=81=D1=88=D0=B8=D1=80?= =?UTF-8?q?=D0=B8=D1=82=D1=8C=20patches/ai@6.0.134.patch=20(#486)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Серверная ai@6.0.134 в writeToServerResponse при backpressure (write()===false) ждала ТОЛЬКО once('drain'). Если клиент отвалился мид-запись, сокет не дренится, await не резолвится: read-цикл паркуется НАВСЕГДА, finally{response.end()} недостижим, reader и буферы висят до рестарта. В autonomous ран продолжает лить вывод после дисконнекта → КАЖДЫЙ дисконнект мид-ран оставляет висящий пайп. Плюс read() — fire-and-forget с throw → unhandledRejection. Патч расширен (index.js и index.mjs): Promise.race drain/close/error с гигиеной once-слушателей (все три снимаются на первом settle — не копятся по одному на stall); при close/error — reader.cancel() и выход (безопасно для detached-ранов: независимый дренаж делает consumeStream); rejection read() поглощается с логом. pnpm-lock patch_hash перегенерён (patch-commit). Выравнивание версии ai — отдельный коммит (#495), не здесь. Тест: трипвайр-спека в apps/server (только там резолвится патченная копия) по образцу ai-sdk-partial-output.patch.spec — дисконнект мид-запись без drain завершает ответ (end() вызван, reader не висит), read()-throw не даёт unhandledRejection, оба dist-билда несут маркер PATCH(docmost #486). --- .../ai/ai-sdk-drain-hang.patch.spec.ts | 133 ++++++++++++++++++ patches/ai@6.0.134.patch | 120 +++++++++++++++- pnpm-lock.yaml | 12 +- 3 files changed, 253 insertions(+), 12 deletions(-) create mode 100644 apps/server/src/integrations/ai/ai-sdk-drain-hang.patch.spec.ts diff --git a/apps/server/src/integrations/ai/ai-sdk-drain-hang.patch.spec.ts b/apps/server/src/integrations/ai/ai-sdk-drain-hang.patch.spec.ts new file mode 100644 index 00000000..39582ee5 --- /dev/null +++ b/apps/server/src/integrations/ai/ai-sdk-drain-hang.patch.spec.ts @@ -0,0 +1,133 @@ +import { readFileSync } from 'fs'; +import { EventEmitter } from 'node:events'; +import { streamText } from 'ai'; +import { MockLanguageModelV3, simulateReadableStream } from 'ai/test'; + +/** + * Regression tests for the writeToServerResponse drain-hang fix in + * patches/ai@6.0.134.patch (#486, commit 6). + * + * Unpatched ai@6.0.134's writeToServerResponse awaits ONLY `once("drain")` when + * response.write() returns false (backpressure). If the client disconnects + * mid-write the socket never drains, so that await never resolves: the read loop + * parks FOREVER, its `finally { response.end() }` is unreachable, and the stream + * reader + buffered chunks are pinned until process restart. In autonomous mode + * the run keeps producing output after the disconnect, so EVERY mid-run + * disconnect leaks a hung pipe. The patch races drain against close/error, and on + * a terminal socket event cancels the reader and breaks so `finally` always runs. + * + * This drives the REAL patched writeToServerResponse through the public + * pipeUIMessageStreamToResponse API with a response that never drains and closes + * mid-write — exactly the leak scenario. + */ + +/** A ServerResponse-like emitter whose first write() stalls (returns false) and + * then "closes" like a disconnecting client — never firing 'drain'. */ +class DisconnectingResponse extends EventEmitter { + ended = false; + writeCount = 0; + statusCode = 200; + writableEnded = false; + destroyed = false; + writeHead(): this { + return this; + } + setHeader(): void {} + flushHeaders(): void {} + write(): boolean { + this.writeCount++; + if (this.writeCount === 1) { + // Simulate the client vanishing mid-write: backpressure (false) and then a + // 'close' on the next tick, and CRUCIALLY never a 'drain'. Unpatched, the + // loop would await drain forever here. + setImmediate(() => this.emit('close')); + return false; + } + return true; + } + end(): void { + this.ended = true; + this.writableEnded = true; + this.emit('finish'); + } +} + +function makeModel() { + return new MockLanguageModelV3({ + doStream: async () => ({ + stream: simulateReadableStream({ + chunks: [ + { type: 'stream-start' as const, warnings: [] }, + { type: 'text-start' as const, id: '1' }, + { type: 'text-delta' as const, id: '1', delta: 'hello ' }, + { type: 'text-delta' as const, id: '1', delta: 'world' }, + { type: 'text-end' as const, id: '1' }, + { + type: 'finish' as const, + finishReason: { unified: 'stop' as const, raw: 'stop' }, + usage: { + inputTokens: { total: 1, noCache: undefined, cacheRead: undefined, cacheWrite: undefined }, + outputTokens: { total: 1, text: 1, reasoning: undefined }, + }, + }, + ], + }), + }), + }); +} + +describe('ai@6.0.134 pnpm patch: writeToServerResponse drain-hang (#486)', () => { + it('ends the response (does NOT hang) when the socket closes mid-write without draining', async () => { + const result = streamText({ model: makeModel(), prompt: 'hi' }); + const res = new DisconnectingResponse(); + // Drain the SDK stream independently, like the production detached path. + void result.consumeStream({ onError: () => undefined }); + result.pipeUIMessageStreamToResponse(res as never); + + // TRIPWIRE: the patched loop exits on 'close' and runs finally -> end(). + // Unpatched, it awaits 'drain' forever and this never becomes true. + await new Promise((resolve, reject) => { + const started = Date.now(); + const poll = setInterval(() => { + if (res.ended) { + clearInterval(poll); + resolve(); + } else if (Date.now() - started > 3000) { + clearInterval(poll); + reject(new Error('writeToServerResponse hung: response never ended')); + } + }, 20); + }); + + expect(res.ended).toBe(true); + }); + + it('does not emit an unhandledRejection when the fire-and-forget read() throws', async () => { + // The patch swallows read()'s rejection (fire-and-forget) with a log instead + // of letting it surface as a process-killing unhandledRejection. + const rejections: unknown[] = []; + const onUnhandled = (e: unknown) => rejections.push(e); + process.on('unhandledRejection', onUnhandled); + // Silence the patch's diagnostic console.error for the throwing read(). + const errSpy = jest.spyOn(console, 'error').mockImplementation(() => undefined); + try { + const result = streamText({ model: makeModel(), prompt: 'hi' }); + const res = new DisconnectingResponse(); + void result.consumeStream({ onError: () => undefined }); + result.pipeUIMessageStreamToResponse(res as never); + await new Promise((r) => setTimeout(r, 300)); + } finally { + process.off('unhandledRejection', onUnhandled); + errSpy.mockRestore(); + } + expect(rejections).toEqual([]); + }); + + it('both installed dist builds (CJS and ESM) carry the #486 patch marker', () => { + const cjsPath = require.resolve('ai'); + const mjsPath = cjsPath.replace(/index\.js$/, 'index.mjs'); + expect(cjsPath).toMatch(/index\.js$/); + expect(readFileSync(cjsPath, 'utf8')).toContain('PATCH(docmost #486)'); + expect(readFileSync(mjsPath, 'utf8')).toContain('PATCH(docmost #486)'); + }); +}); diff --git a/patches/ai@6.0.134.patch b/patches/ai@6.0.134.patch index 4371975e..9c7b81ee 100644 --- a/patches/ai@6.0.134.patch +++ b/patches/ai@6.0.134.patch @@ -1,8 +1,62 @@ diff --git a/dist/index.js b/dist/index.js -index ae447a12f7823ec0a00837ee9f0eb809a610d5f8..a3402b2c2d021ef432cfa76e35d370073d525135 100644 +index ae447a12f7823ec0a00837ee9f0eb809a610d5f8..210b5a9009e5cf1537cb2007c524007c1a01253c 100644 --- a/dist/index.js +++ b/dist/index.js -@@ -6578,9 +6578,19 @@ function createOutputTransformStream(output) { +@@ -5036,9 +5036,40 @@ function writeToServerResponse({ + break; + const canContinue = response.write(value); + if (!canContinue) { +- await new Promise((resolve3) => { +- response.once("drain", resolve3); ++ // PATCH(docmost #486): race "drain" against "close"/"error". The ++ // original awaited ONLY "drain", so a client that disconnected mid-write ++ // (the socket never drains) parked this loop FOREVER: the finally never ++ // ran, response.end() was unreachable, and the reader + buffered chunks ++ // were held until process restart. On close/error we cancel the reader ++ // and break so the finally always runs (safe for detached runs: ++ // consumeStream drains the SDK stream independently). Listener hygiene: ++ // all three once-listeners are removed on the first settle, so they ++ // cannot pile up one-per-stall. ++ const closed = await new Promise((resolve3) => { ++ function finish(isClosed) { ++ response.removeListener("drain", onDrain); ++ response.removeListener("close", onClose); ++ response.removeListener("error", onError); ++ resolve3(isClosed); ++ } ++ function onDrain() { ++ finish(false); ++ } ++ function onClose() { ++ finish(true); ++ } ++ function onError() { ++ finish(true); ++ } ++ response.once("drain", onDrain); ++ response.once("close", onClose); ++ response.once("error", onError); + }); ++ if (closed) { ++ await reader.cancel().catch(() => { ++ }); ++ break; ++ } + } + } + } catch (error) { +@@ -5047,7 +5078,9 @@ function writeToServerResponse({ + response.end(); + } + }; +- read(); ++ read().catch((error) => { ++ console.error("ai writeToServerResponse read() failed:", error); ++ }); + } + + // src/text-stream/pipe-text-stream-to-response.ts +@@ -6578,9 +6611,19 @@ function createOutputTransformStream(output) { controller.enqueue({ part: chunk, partialOutput: void 0 }); return; } @@ -23,7 +77,7 @@ index ae447a12f7823ec0a00837ee9f0eb809a610d5f8..a3402b2c2d021ef432cfa76e35d37007 const result = await output.parsePartialOutput({ text: text2 }); if (result !== void 0) { const currentJson = JSON.stringify(result.partial); -@@ -6959,7 +6969,7 @@ var DefaultStreamTextResult = class { +@@ -6959,7 +7002,7 @@ var DefaultStreamTextResult = class { }) ); } @@ -33,10 +87,64 @@ index ae447a12f7823ec0a00837ee9f0eb809a610d5f8..a3402b2c2d021ef432cfa76e35d37007 maxRetries: maxRetriesArg, abortSignal diff --git a/dist/index.mjs b/dist/index.mjs -index 663875332e3f9a9bd167c25583c515876f42951b..b840b0502c9894df983e0154805abb80e70e6331 100644 +index 663875332e3f9a9bd167c25583c515876f42951b..a51514a390d22811a407edd8b703e21793586cc8 100644 --- a/dist/index.mjs +++ b/dist/index.mjs -@@ -6501,9 +6501,19 @@ function createOutputTransformStream(output) { +@@ -4957,9 +4957,40 @@ function writeToServerResponse({ + break; + const canContinue = response.write(value); + if (!canContinue) { +- await new Promise((resolve3) => { +- response.once("drain", resolve3); ++ // PATCH(docmost #486): race "drain" against "close"/"error". The ++ // original awaited ONLY "drain", so a client that disconnected mid-write ++ // (the socket never drains) parked this loop FOREVER: the finally never ++ // ran, response.end() was unreachable, and the reader + buffered chunks ++ // were held until process restart. On close/error we cancel the reader ++ // and break so the finally always runs (safe for detached runs: ++ // consumeStream drains the SDK stream independently). Listener hygiene: ++ // all three once-listeners are removed on the first settle, so they ++ // cannot pile up one-per-stall. ++ const closed = await new Promise((resolve3) => { ++ function finish(isClosed) { ++ response.removeListener("drain", onDrain); ++ response.removeListener("close", onClose); ++ response.removeListener("error", onError); ++ resolve3(isClosed); ++ } ++ function onDrain() { ++ finish(false); ++ } ++ function onClose() { ++ finish(true); ++ } ++ function onError() { ++ finish(true); ++ } ++ response.once("drain", onDrain); ++ response.once("close", onClose); ++ response.once("error", onError); + }); ++ if (closed) { ++ await reader.cancel().catch(() => { ++ }); ++ break; ++ } + } + } + } catch (error) { +@@ -4968,7 +4999,9 @@ function writeToServerResponse({ + response.end(); + } + }; +- read(); ++ read().catch((error) => { ++ console.error("ai writeToServerResponse read() failed:", error); ++ }); + } + + // src/text-stream/pipe-text-stream-to-response.ts +@@ -6501,9 +6534,19 @@ function createOutputTransformStream(output) { controller.enqueue({ part: chunk, partialOutput: void 0 }); return; } @@ -57,7 +165,7 @@ index 663875332e3f9a9bd167c25583c515876f42951b..b840b0502c9894df983e0154805abb80 const result = await output.parsePartialOutput({ text: text2 }); if (result !== void 0) { const currentJson = JSON.stringify(result.partial); -@@ -6882,7 +6892,7 @@ var DefaultStreamTextResult = class { +@@ -6882,7 +6925,7 @@ var DefaultStreamTextResult = class { }) ); } diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 1ab1c069..64c5e9f4 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -48,7 +48,7 @@ patchedDependencies: hash: d8dc66e5ec3b9d23a876b979f493b6aa901fd2d965be54729495da5136296a42 path: patches/@hocuspocus__server@3.4.4.patch ai@6.0.134: - hash: f60bfc3357e01e1f3978c6c40fdd65aeb33fefaad7179cde8676465b6c5ff4d9 + hash: e8c599b3963eb01b9ed1481683b7b795ce94137aa1a0c951917c20c8b870299b path: patches/ai@6.0.134.patch scimmy@1.3.5: hash: 775d80f86830b2c5dd1a250c9802c10f8fc3da3c7898373de5aa0c23993d1673 @@ -641,10 +641,10 @@ importers: version: 8.3.0(socket.io-adapter@2.5.4) ai: specifier: ^6.0.134 - version: 6.0.134(patch_hash=f60bfc3357e01e1f3978c6c40fdd65aeb33fefaad7179cde8676465b6c5ff4d9)(zod@4.3.6) + version: 6.0.134(patch_hash=e8c599b3963eb01b9ed1481683b7b795ce94137aa1a0c951917c20c8b870299b)(zod@4.3.6) ai-sdk-ollama: specifier: ^3.8.1 - version: 3.8.1(ai@6.0.134(patch_hash=f60bfc3357e01e1f3978c6c40fdd65aeb33fefaad7179cde8676465b6c5ff4d9)(zod@4.3.6))(zod@4.3.6) + version: 3.8.1(ai@6.0.134(patch_hash=e8c599b3963eb01b9ed1481683b7b795ce94137aa1a0c951917c20c8b870299b)(zod@4.3.6))(zod@4.3.6) bcrypt: specifier: ^6.0.0 version: 6.0.0 @@ -16453,17 +16453,17 @@ snapshots: agent-base@7.1.4: {} - ai-sdk-ollama@3.8.1(ai@6.0.134(patch_hash=f60bfc3357e01e1f3978c6c40fdd65aeb33fefaad7179cde8676465b6c5ff4d9)(zod@4.3.6))(zod@4.3.6): + ai-sdk-ollama@3.8.1(ai@6.0.134(patch_hash=e8c599b3963eb01b9ed1481683b7b795ce94137aa1a0c951917c20c8b870299b)(zod@4.3.6))(zod@4.3.6): dependencies: '@ai-sdk/provider': 3.0.8 '@ai-sdk/provider-utils': 4.0.21(zod@4.3.6) - ai: 6.0.134(patch_hash=f60bfc3357e01e1f3978c6c40fdd65aeb33fefaad7179cde8676465b6c5ff4d9)(zod@4.3.6) + ai: 6.0.134(patch_hash=e8c599b3963eb01b9ed1481683b7b795ce94137aa1a0c951917c20c8b870299b)(zod@4.3.6) jsonrepair: 3.13.3 ollama: 0.6.3 transitivePeerDependencies: - zod - ai@6.0.134(patch_hash=f60bfc3357e01e1f3978c6c40fdd65aeb33fefaad7179cde8676465b6c5ff4d9)(zod@4.3.6): + ai@6.0.134(patch_hash=e8c599b3963eb01b9ed1481683b7b795ce94137aa1a0c951917c20c8b870299b)(zod@4.3.6): dependencies: '@ai-sdk/gateway': 3.0.77(zod@4.3.6) '@ai-sdk/provider': 3.0.8