From 6d27ad17649f9013991b69ef1a6534b685ad226e Mon Sep 17 00:00:00 2001 From: Peter Wielander Date: Mon, 22 Jun 2026 11:22:05 -0700 Subject: [PATCH] Revert "fix(world-vercel): cancel v4 event frame stream on early exit (#2547)" This reverts commit e3672e84f4996fbd35fc2542d11d401979d76924. --- .changeset/cancel-v4-frame-stream.md | 5 -- packages/world-vercel/src/events-v4.test.ts | 47 ------------ packages/world-vercel/src/frames.test.ts | 70 +----------------- packages/world-vercel/src/frames.ts | 80 ++++++++------------- 4 files changed, 31 insertions(+), 171 deletions(-) delete mode 100644 .changeset/cancel-v4-frame-stream.md diff --git a/.changeset/cancel-v4-frame-stream.md b/.changeset/cancel-v4-frame-stream.md deleted file mode 100644 index 9383ae3cc4..0000000000 --- a/.changeset/cancel-v4-frame-stream.md +++ /dev/null @@ -1,5 +0,0 @@ ---- -'@workflow/world-vercel': patch ---- - -Cancel the v4 event frame stream when a reader stops early, so the response body's undici connection returns to the pool instead of leaking. diff --git a/packages/world-vercel/src/events-v4.test.ts b/packages/world-vercel/src/events-v4.test.ts index 81c54739b5..9bca92a305 100644 --- a/packages/world-vercel/src/events-v4.test.ts +++ b/packages/world-vercel/src/events-v4.test.ts @@ -10,7 +10,6 @@ import { MockAgent } from 'undici'; import { afterEach, describe, expect, it, vi } from 'vitest'; import { createWorkflowRunEventV4, - getEventV4, getWorkflowRunEventsV4, throwForErrorResponse, } from './events-v4.js'; @@ -237,52 +236,6 @@ describe('getWorkflowRunEventsV4 over HTTP', () => { }); }); -/** - * getEventV4 returns after the first frame. The early return must cancel the - * response body (releasing its undici socket) without corrupting the returned - * value or hanging — the trailing frame below is never read. - */ -describe('getEventV4 over HTTP', () => { - it('returns the first frame and stops reading the rest', async () => { - const origin = 'https://vercel-workflow.com'; - const agent = new MockAgent(); - agent.disableNetConnect(); - - const body = new TextEncoder().encode('event-payload'); - const frames = Buffer.concat([ - encodeFrame( - { - eventId: 'evnt_1', - runId: 'wrun_1', - eventType: 'run_created', - createdAt: '2026-06-10T00:00:00.000Z', - eventData: {}, - }, - body - ), - // Trailing bytes the reader must never need. - encodeFrame({ eventId: 'evnt_unused' }, new Uint8Array(8)), - ]); - - agent - .get(origin) - .intercept({ path: '/api/v4/runs/wrun_1/events/evnt_1', method: 'GET' }) - .reply(200, frames, { - headers: { 'content-type': V4_FRAME_CONTENT_TYPE }, - }); - - const { event, body: returnedBody } = await getEventV4('wrun_1', 'evnt_1', { - token: 'test-token', - dispatcher: agent, - }); - - expect(event.eventId).toBe('evnt_1'); - expect(event.eventType).toBe('run_created'); - expect(new Uint8Array(returnedBody)).toEqual(body); - agent.assertNoPendingInterceptors(); - }); -}); - /** * Regression: v4 requests must go through the global `fetch`, not undici's * `request()`. Vercel's observability log viewer instruments the global diff --git a/packages/world-vercel/src/frames.test.ts b/packages/world-vercel/src/frames.test.ts index 72d312272e..966bab97ab 100644 --- a/packages/world-vercel/src/frames.test.ts +++ b/packages/world-vercel/src/frames.test.ts @@ -1,4 +1,4 @@ -import { decode } from 'cbor-x'; +import { decode, encode } from 'cbor-x'; import { describe, expect, it } from 'vitest'; import { type DecodedFrame, @@ -41,29 +41,6 @@ async function drainFrames( return out; } -/** A stream that stays open after delivering its payload (never signals EOF), - * like a kept-alive HTTP socket, and records whether cancel() ran — the - * signal undici uses to release the connection. highWaterMark: 0 suppresses - * the pull-ahead that would otherwise auto-close a toy stream. */ -function spyStream(payload: Uint8Array) { - let sent = false; - let cancelled = false; - const stream = new ReadableStream( - { - pull(controller) { - if (sent) return; - controller.enqueue(payload); - sent = true; - }, - cancel() { - cancelled = true; - }, - }, - { highWaterMark: 0 } - ); - return { stream, wasCancelled: () => cancelled }; -} - describe('encodeFrame', () => { it('produces the canonical wire layout', () => { const meta = { eventId: 'evnt_abc', n: 42 }; @@ -233,51 +210,6 @@ describe('decodeFrames from an AsyncIterable source', () => { }); }); -describe('decodeFrames releases the stream on early exit', () => { - // Regression: a consumer that stops before EOF (getEventV4 returns after - // the first frame; consumeListFrameStream breaks at the sentinel) must - // cancel the body, or its undici socket stays pinned out of the pool. - function twoFramesThenEnd(): Uint8Array { - return new Uint8Array([ - ...encodeFrame({ eventId: 'a' }, new Uint8Array([1, 2, 3])), - ...encodeFrame({ eventId: 'b' }, new Uint8Array([4, 5, 6])), - ...encodeEndFrame(), - ]); - } - - it('cancels the underlying stream when the consumer breaks early', async () => { - const { stream, wasCancelled } = spyStream(twoFramesThenEnd()); - for await (const f of decodeFrames(stream)) { - expect(f.meta).toEqual({ eventId: 'a' }); - break; // mirrors getEventV4 returning after the first frame - } - expect(wasCancelled()).toBe(true); - }); - - it('cancels via the reader path when the source is not async-iterable', async () => { - const { stream, wasCancelled } = spyStream(twoFramesThenEnd()); - // A bare { getReader } object forces the readerToIterator branch (a real - // ReadableStream is already async-iterable in Node). - const source = { - getReader: () => stream.getReader(), - } as unknown as ReadableStream; - for await (const f of decodeFrames(source)) { - expect(f.meta).toEqual({ eventId: 'a' }); - break; - } - expect(wasCancelled()).toBe(true); - }); - - it('still decodes every frame when fully consumed', async () => { - const frames = await drainFrames(spyStream(twoFramesThenEnd()).stream); - expect(frames.map((f) => f.meta)).toEqual([ - { eventId: 'a' }, - { eventId: 'b' }, - { _end: 1 }, - ]); - }); -}); - describe('V4_FRAME_CONTENT_TYPE', () => { it('matches the server-side content type', () => { expect(V4_FRAME_CONTENT_TYPE).toBe('application/vnd.workflow.v4-frames'); diff --git a/packages/world-vercel/src/frames.ts b/packages/world-vercel/src/frames.ts index 2d802063bc..59aacd9280 100644 --- a/packages/world-vercel/src/frames.ts +++ b/packages/world-vercel/src/frames.ts @@ -79,56 +79,41 @@ export async function* decodeFrames( return out; }; - try { - while (true) { - if (!(await refill(4))) return; - const metaLen = new DataView( - buffer.buffer, - buffer.byteOffset, - 4 - ).getUint32(0, false); - take(4); + while (true) { + if (!(await refill(4))) return; + const metaLen = new DataView(buffer.buffer, buffer.byteOffset, 4).getUint32( + 0, + false + ); + take(4); - if (!(await refill(metaLen))) { - throw new Error('decodeFrames: truncated meta block'); - } - const meta = decode(take(metaLen)) as Record; + if (!(await refill(metaLen))) { + throw new Error('decodeFrames: truncated meta block'); + } + const meta = decode(take(metaLen)) as Record; - if (!(await refill(4))) { - throw new Error('decodeFrames: truncated body length'); - } - const bodyLen = new DataView( - buffer.buffer, - buffer.byteOffset, - 4 - ).getUint32(0, false); - take(4); + if (!(await refill(4))) { + throw new Error('decodeFrames: truncated body length'); + } + const bodyLen = new DataView(buffer.buffer, buffer.byteOffset, 4).getUint32( + 0, + false + ); + take(4); - if (bodyLen > 0 && !(await refill(bodyLen))) { + if (bodyLen > 0) { + if (!(await refill(bodyLen))) { throw new Error('decodeFrames: truncated body bytes'); } - // Slice (not subarray) so the yielded body owns its bytes — later - // reads into the buffer won't overwrite it; bodyLen 0 yields empty. + // Slice (not subarray) so the yielded body owns its bytes — + // subsequent reads into the buffer won't overwrite it. yield { meta, body: buffer.slice(0, bodyLen) }; take(bodyLen); - - if (meta._end === 1) return; + } else { + yield { meta, body: new Uint8Array(0) }; } - } finally { - // Release the source when the consumer stops before EOF (early - // break/return): an unconsumed body pins its undici socket out of the - // connection pool. No-op once the stream is already drained. - await closeQuietly(() => chunks.return?.()); - } -} -/** Best-effort source cleanup, safe to run from a `finally`: swallows errors - * so cleanup can't mask the original outcome. */ -async function closeQuietly(close: () => unknown): Promise { - try { - await close(); - } catch { - // best-effort + if (meta._end === 1) return; } } @@ -137,14 +122,9 @@ async function closeQuietly(close: () => unknown): Promise { async function* readerToIterator( reader: ReadableStreamDefaultReader ): AsyncGenerator { - try { - while (true) { - const { done, value } = await reader.read(); - if (done) return; - if (value) yield value; - } - } finally { - // Cancel on early exit so the socket is released, not just unlocked. - await closeQuietly(() => reader.cancel()); + while (true) { + const { done, value } = await reader.read(); + if (done) return; + if (value) yield value; } }