From 9e82aff0e3b576e66dcca6488714c7eddbc1aed6 Mon Sep 17 00:00:00 2001 From: Rajeh Taher Date: Fri, 18 Sep 2026 14:29:14 +0200 Subject: [PATCH 1/5] feat(cli): add event stream over the public event stream --- README.md | 27 ++++ src/commands/event/stream.ts | 129 ++++++++++++++++ test/commands/event-stream.test.ts | 237 +++++++++++++++++++++++++++++ 3 files changed, 393 insertions(+) create mode 100644 src/commands/event/stream.ts create mode 100644 test/commands/event-stream.test.ts diff --git a/README.md b/README.md index 078d5d42..90745e92 100644 --- a/README.md +++ b/README.md @@ -1064,6 +1064,33 @@ that includes the current operation. response excerpt that the API records for a failed HTTP response. Human output escapes control characters in remote text. +### Stream events + +`polymorfa event stream` follows a project's public event stream and prints +each event as it happens. It needs a credential with `events:listen` and a team +enrolled in the Event streams beta; otherwise it exits with the authorization +code. + +```shell +polymorfa event stream \ + --owner project \ + --project project_123 \ + --type 'message.*' \ + --json +``` + +`--json` writes one JSON line per event with its `cursor`, the event metadata, +and the decoded webhook body (`null` when hosted message storage did not keep +it). Human output prints the time, type, ID, and body, and prints a +`--from ` hint on exit. Pass `--from` to resume within 24 hours. The +command reconnects with backoff after a dropped connection. If events after +`--from` expired before delivery, it reports the count and continues; use +`--on-gap stop` to exit with the stream gap code instead. `--max-events` stops +after a fixed number of events. + +Use `event stream` to consume events in a script. Use `listen` to forward exact +bodies to a local webhook handler. + ## Local event listening and forwarding `polymorfa listen` opens the project listener stream and can forward the exact diff --git a/src/commands/event/stream.ts b/src/commands/event/stream.ts new file mode 100644 index 00000000..3c909a3a --- /dev/null +++ b/src/commands/event/stream.ts @@ -0,0 +1,129 @@ +import {Flags} from '@oclif/core' +import type {EventStreamItem} from '@polymorfa/sdk' + +import {BaseCommand} from '../../base-command.js' +import {CliConditionError} from '../../errors/conditions.js' + +export default class EventStream extends BaseCommand { + static override description = + "Stream a project's events as they happen, with automatic reconnect and resume. Requires events:listen and the Event streams beta." + + static override examples = [ + '<%= config.bin %> event stream --owner project --project 11111111-1111-4111-8111-111111111111 --type "message.*"', + '<%= config.bin %> event stream --owner project --project 11111111-1111-4111-8111-111111111111 --from "$CURSOR" --json', + ] + + static override flags = { + from: Flags.string({ + description: + 'Resume after this cursor from a previous run; valid for 24 hours.', + }), + 'max-events': Flags.integer({ + description: 'Stop after this many events.', + min: 1, + }), + 'on-gap': Flags.string({ + default: 'continue', + description: + 'Continue or stop when events after --from expired before delivery.', + options: ['continue', 'stop'], + }), + type: Flags.string({ + description: + 'Stream an event type or trailing-wildcard pattern such as message.*.', + multiple: true, + }), + } + + async run(): Promise { + const {flags} = await this.parse(EventStream) + const session = await this.developerSession(flags) + if (session.context.owner !== 'project') { + throw new CliConditionError( + 'action_required', + 'Event streams belong to a project. Pass --owner project and --project.', + ) + } + const json = flags.json === true + const quiet = flags.quiet === true + const controller = new AbortController() + const stop = () => controller.abort() + process.once('SIGINT', stop) + process.once('SIGTERM', stop) + let gap: {missedEvents: number} | undefined + let delivered = 0 + let lastCursor: string | undefined = flags.from + try { + const stream = session.client.events.stream({ + ...(flags.type === undefined ? {} : {types: flags.type}), + ...(flags.from === undefined ? {} : {since: flags.from}), + signal: controller.signal, + onGap: (value) => { + gap = value + if (json) { + process.stdout.write( + `${JSON.stringify({ + schemaVersion: 1, + type: 'gap', + reason: value.reason, + missedEvents: value.missedEvents, + requestedCursor: value.requestedCursor, + })}\n`, + ) + } else { + process.stderr.write( + `${value.missedEvents} events after the resume cursor expired before delivery. Use "event list" to read them.\n`, + ) + } + if (flags['on-gap'] === 'stop') controller.abort() + }, + onReconnect: (_error, delayMs) => { + if (!json && !quiet) { + process.stderr.write(`Reconnecting in ${delayMs} ms...\n`) + } + }, + }) + for await (const item of stream) { + delivered += 1 + lastCursor = item.cursor + if (json) process.stdout.write(`${JSON.stringify(jsonLine(item))}\n`) + else if (!quiet) process.stdout.write(`${humanLine(item)}\n`) + if ( + flags['max-events'] !== undefined && + delivered >= flags['max-events'] + ) + break + } + } finally { + process.removeListener('SIGINT', stop) + process.removeListener('SIGTERM', stop) + if (!json && !quiet && lastCursor !== undefined) { + process.stderr.write(`Resume with --from ${lastCursor}\n`) + } + } + if (gap !== undefined && flags['on-gap'] === 'stop') { + throw new CliConditionError( + 'stream_gap', + `${gap.missedEvents} events expired before delivery; the stream stopped.`, + ) + } + } +} + +function jsonLine(item: EventStreamItem) { + return { + schemaVersion: 1, + type: 'event', + cursor: item.cursor, + event: {...item.event, payload: undefined}, + webhook: item.webhook, + } +} + +function humanLine(item: EventStreamItem): string { + const body = + item.webhook === null + ? `(no body: ${item.event.payloadAvailability})` + : JSON.stringify(item.webhook.payload) + return `${item.event.createdAt}\t${item.event.type}\t${item.event.id}\t${body}` +} diff --git a/test/commands/event-stream.test.ts b/test/commands/event-stream.test.ts new file mode 100644 index 00000000..8f0f3871 --- /dev/null +++ b/test/commands/event-stream.test.ts @@ -0,0 +1,237 @@ +import {afterEach, beforeEach, describe, expect, test, vi} from 'vitest' + +import EventStream from '../../src/commands/event/stream.js' +import {withEnvironment} from '../helpers/http-server.js' + +const PROJECT = '11111111-1111-4111-8111-111111111111' + +const sdk = vi.hoisted(() => ({ + params: [] as Array>, + items: [] as unknown[], + gap: undefined as {missedEvents: number} | undefined, + failure: undefined as Error | undefined, +})) + +vi.mock('@polymorfa/sdk', async (importOriginal) => { + const original = await importOriginal() + + class Client { + readonly events = { + stream: (params: Record) => { + sdk.params.push(params) + return { + async *[Symbol.asyncIterator]() { + await Promise.resolve() + if (sdk.gap !== undefined) { + ;(params.onGap as (gap: unknown) => void)({ + reason: 'retention_exceeded', + missedEvents: sdk.gap.missedEvents, + requestedCursor: 'c0', + }) + } + for (const item of sdk.items) { + if ((params.signal as AbortSignal).aborted) return + yield item + } + if (sdk.failure !== undefined) throw sdk.failure + }, + } + }, + } + + project(): this { + return this + } + } + + return {...original, Client} +}) + +describe('event stream', () => { + beforeEach(() => { + sdk.params.length = 0 + sdk.items = [item('e1', 'c1'), item('e2', 'c2', false)] + sdk.gap = undefined + sdk.failure = undefined + }) + + afterEach(() => vi.restoreAllMocks()) + + test('streams JSON lines with filters and a resume cursor', async () => { + const output = vi + .spyOn(process.stdout, 'write') + .mockImplementation(() => true) + await environment(() => + EventStream.run([ + '--owner', + 'project', + '--project', + PROJECT, + '--type', + 'message.*', + '--type', + 'session.connected', + '--from', + 'c0', + '--json', + ]), + ) + expect(sdk.params[0]).toMatchObject({ + types: ['message.*', 'session.connected'], + since: 'c0', + }) + const lines = output.mock.calls.map( + (call) => JSON.parse(String(call[0])) as Record, + ) + expect(lines).toHaveLength(2) + expect(lines[0]).toMatchObject({ + schemaVersion: 1, + type: 'event', + cursor: 'c1', + event: {id: 'e1', type: 'message.received'}, + webhook: {id: 'e1'}, + }) + expect(lines[1]).toMatchObject({ + schemaVersion: 1, + type: 'event', + cursor: 'c2', + event: {id: 'e2'}, + webhook: null, + }) + expect(lines[0]?.event).not.toHaveProperty('payload') + }) + + test('prints one human line per event and the resume cursor', async () => { + const output = vi + .spyOn(process.stdout, 'write') + .mockImplementation(() => true) + const errors = vi + .spyOn(process.stderr, 'write') + .mockImplementation(() => true) + await environment(() => + EventStream.run(['--owner', 'project', '--project', PROJECT]), + ) + expect(String(output.mock.calls[0]?.[0])).toBe( + '2026-09-18T10:00:00.000Z\tmessage.received\te1\t{"text":"hi"}\n', + ) + expect(String(output.mock.calls[1]?.[0])).toContain( + '(no body: not_retained)', + ) + expect(errors.mock.calls.map((call) => String(call[0]))).toContain( + 'Resume with --from c2\n', + ) + }) + + test('stops after --max-events', async () => { + const output = vi + .spyOn(process.stdout, 'write') + .mockImplementation(() => true) + await environment(() => + EventStream.run([ + '--owner', + 'project', + '--project', + PROJECT, + '--max-events', + '1', + '--json', + ]), + ) + expect(output).toHaveBeenCalledTimes(1) + }) + + test('exits with the stream gap code when --on-gap stop sees expired events', async () => { + sdk.gap = {missedEvents: 3} + vi.spyOn(process.stdout, 'write').mockImplementation(() => true) + vi.spyOn(process.stderr, 'write').mockImplementation(() => true) + const error = await environment(() => + EventStream.run([ + '--owner', + 'project', + '--project', + PROJECT, + '--from', + 'c0', + '--on-gap', + 'stop', + '--json', + ]), + ).catch((caught: unknown) => caught) + expect(exitCode(error)).toBe(9) + }) + + test('requires a project owner', async () => { + vi.spyOn(process.stderr, 'write').mockImplementation(() => true) + const error = await environment(() => + EventStream.run(['--owner', 'organization', '--json']), + ).catch((caught: unknown) => caught) + expect(exitCode(error)).toBe(8) + expect(sdk.params).toHaveLength(0) + }) + + test('maps a disabled beta to the authorization exit code', async () => { + const {PolymorfaAuthorizationError} = await import('@polymorfa/sdk') + sdk.items = [] + sdk.failure = new PolymorfaAuthorizationError( + 'Event streams is not enabled for this team.', + {code: 'feature_unavailable', status: 403}, + ) + vi.spyOn(process.stderr, 'write').mockImplementation(() => true) + const error = await environment(() => + EventStream.run(['--owner', 'project', '--project', PROJECT, '--json']), + ).catch((caught: unknown) => caught) + expect(exitCode(error)).toBe(3) + }) +}) + +function item(id: string, cursor: string, withBody = true) { + return { + cursor, + streamId: 's', + sequence: 2, + event: { + id, + organizationId: 'org_1', + projectId: PROJECT, + type: 'message.received', + source: 'runtime', + environment: 'development', + createdAt: '2026-09-18T10:00:00.000Z', + payloadAvailability: withBody ? 'available' : 'not_retained', + payload: withBody + ? {encoding: 'base64', contentType: 'application/json', data: 'e30='} + : null, + replayableUntil: null, + metadataExpiresAt: '2026-10-18T10:00:00.000Z', + }, + webhook: withBody + ? { + id, + session: 'support', + timestamp: '2026-09-18T10:00:00.000Z', + event: 'message.received', + payload: {text: 'hi'}, + } + : null, + } +} + +function environment(run: () => Promise): Promise { + return withEnvironment( + { + POLYMORFA_API_KEY: + 'pmfa_AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA', + POLYMORFA_BASE_URL: 'https://api.example.test', + POLYMORFA_CLIENT_TOKEN: undefined, + POLYMORFA_PROJECT_TOKEN: undefined, + }, + run, + ) +} + +function exitCode(error: unknown): number | undefined { + if (typeof error !== 'object' || error === null || !('oclif' in error)) { + return undefined + } + return (error as {oclif?: {exit?: number}}).oclif?.exit +} From 95acc22d14bb7ec969446ec403bfdb1835176078 Mon Sep 17 00:00:00 2001 From: Rajeh Taher Date: Tue, 22 Sep 2026 12:34:06 +0200 Subject: [PATCH 2/5] fix(event): wait for stdout drain while streaming events --- src/commands/event/stream.ts | 14 ++++++++++++-- 1 file changed, 12 insertions(+), 2 deletions(-) diff --git a/src/commands/event/stream.ts b/src/commands/event/stream.ts index 3c909a3a..819a8dcf 100644 --- a/src/commands/event/stream.ts +++ b/src/commands/event/stream.ts @@ -1,3 +1,4 @@ +import {once} from 'node:events' import {Flags} from '@oclif/core' import type {EventStreamItem} from '@polymorfa/sdk' @@ -86,8 +87,8 @@ export default class EventStream extends BaseCommand { for await (const item of stream) { delivered += 1 lastCursor = item.cursor - if (json) process.stdout.write(`${JSON.stringify(jsonLine(item))}\n`) - else if (!quiet) process.stdout.write(`${humanLine(item)}\n`) + if (json) await writeStdout(`${JSON.stringify(jsonLine(item))}\n`) + else if (!quiet) await writeStdout(`${humanLine(item)}\n`) if ( flags['max-events'] !== undefined && delivered >= flags['max-events'] @@ -110,6 +111,15 @@ export default class EventStream extends BaseCommand { } } +/** + * Waits for stdout to accept each line so a slow consumer pauses the stream + * instead of growing the process's output buffer without bound. + */ +async function writeStdout(value: string): Promise { + if (process.stdout.write(value)) return + await once(process.stdout, 'drain') +} + function jsonLine(item: EventStreamItem) { return { schemaVersion: 1, From e35b8f2fc46b73a4ce9adec88d94409a263afc8c Mon Sep 17 00:00:00 2001 From: Rajeh Taher Date: Tue, 22 Sep 2026 13:55:57 +0200 Subject: [PATCH 3/5] fix(event): let SIGINT/SIGTERM interrupt a stdout drain wait --- src/commands/event/stream.ts | 24 +++++++++++++++++++----- test/commands/event-stream.test.ts | 12 ++++++++++++ 2 files changed, 31 insertions(+), 5 deletions(-) diff --git a/src/commands/event/stream.ts b/src/commands/event/stream.ts index 819a8dcf..e0978083 100644 --- a/src/commands/event/stream.ts +++ b/src/commands/event/stream.ts @@ -87,8 +87,14 @@ export default class EventStream extends BaseCommand { for await (const item of stream) { delivered += 1 lastCursor = item.cursor - if (json) await writeStdout(`${JSON.stringify(jsonLine(item))}\n`) - else if (!quiet) await writeStdout(`${humanLine(item)}\n`) + if (json) + await writeStdout( + `${JSON.stringify(jsonLine(item))}\n`, + controller.signal, + ) + else if (!quiet) + await writeStdout(`${humanLine(item)}\n`, controller.signal) + if (controller.signal.aborted) break if ( flags['max-events'] !== undefined && delivered >= flags['max-events'] @@ -113,11 +119,19 @@ export default class EventStream extends BaseCommand { /** * Waits for stdout to accept each line so a slow consumer pauses the stream - * instead of growing the process's output buffer without bound. + * instead of growing the process's output buffer without bound. An abort + * (Ctrl-C or SIGTERM) ends the wait. */ -async function writeStdout(value: string): Promise { +async function writeStdout(value: string, signal: AbortSignal): Promise { if (process.stdout.write(value)) return - await once(process.stdout, 'drain') + try { + // SIGINT/SIGTERM abort the controller; stop waiting so the command exits + // even when the consumer never drains stdout. + await once(process.stdout, 'drain', {signal}) + } catch (error) { + if (signal.aborted) return + throw error + } } function jsonLine(item: EventStreamItem) { diff --git a/test/commands/event-stream.test.ts b/test/commands/event-stream.test.ts index 8f0f3871..b23249a3 100644 --- a/test/commands/event-stream.test.ts +++ b/test/commands/event-stream.test.ts @@ -140,6 +140,18 @@ describe('event stream', () => { expect(output).toHaveBeenCalledTimes(1) }) + test('SIGINT ends a write blocked on stdout backpressure', async () => { + const output = vi.spyOn(process.stdout, 'write').mockImplementation(() => { + setImmediate(() => process.emit('SIGINT', 'SIGINT')) + return false + }) + vi.spyOn(process.stderr, 'write').mockImplementation(() => true) + await environment(() => + EventStream.run(['--owner', 'project', '--project', PROJECT, '--json']), + ) + expect(output).toHaveBeenCalledTimes(1) + }) + test('exits with the stream gap code when --on-gap stop sees expired events', async () => { sdk.gap = {missedEvents: 3} vi.spyOn(process.stdout, 'write').mockImplementation(() => true) From 95d5025d626ff320a3da145a423c4b93ca5d14f3 Mon Sep 17 00:00:00 2001 From: Rajeh Taher Date: Tue, 22 Sep 2026 14:05:08 +0200 Subject: [PATCH 4/5] fix(event): end the stream cleanly when stdout's pipe closes --- src/commands/event/stream.ts | 20 ++++++++++++++++++-- test/commands/event-stream.test.ts | 19 +++++++++++++++++++ 2 files changed, 37 insertions(+), 2 deletions(-) diff --git a/src/commands/event/stream.ts b/src/commands/event/stream.ts index e0978083..db4dcd42 100644 --- a/src/commands/event/stream.ts +++ b/src/commands/event/stream.ts @@ -51,6 +51,13 @@ export default class EventStream extends BaseCommand { const stop = () => controller.abort() process.once('SIGINT', stop) process.once('SIGTERM', stop) + // A consumer that closes the pipe (for example `| head -n 1`) ends the + // stream normally instead of failing the command. + const stdoutClosed = (error: unknown) => { + if (isClosedPipe(error)) controller.abort() + else throw error + } + process.stdout.on('error', stdoutClosed) let gap: {missedEvents: number} | undefined let delivered = 0 let lastCursor: string | undefined = flags.from @@ -104,6 +111,7 @@ export default class EventStream extends BaseCommand { } finally { process.removeListener('SIGINT', stop) process.removeListener('SIGTERM', stop) + process.stdout.removeListener('error', stdoutClosed) if (!json && !quiet && lastCursor !== undefined) { process.stderr.write(`Resume with --from ${lastCursor}\n`) } @@ -120,7 +128,7 @@ export default class EventStream extends BaseCommand { /** * Waits for stdout to accept each line so a slow consumer pauses the stream * instead of growing the process's output buffer without bound. An abort - * (Ctrl-C or SIGTERM) ends the wait. + * (Ctrl-C, SIGTERM or a closed stdout pipe) ends the wait. */ async function writeStdout(value: string, signal: AbortSignal): Promise { if (process.stdout.write(value)) return @@ -129,11 +137,19 @@ async function writeStdout(value: string, signal: AbortSignal): Promise { // even when the consumer never drains stdout. await once(process.stdout, 'drain', {signal}) } catch (error) { - if (signal.aborted) return + if (signal.aborted || isClosedPipe(error)) return throw error } } +function isClosedPipe(error: unknown): boolean { + return ( + error instanceof Error && + 'code' in error && + (error.code === 'EPIPE' || error.code === 'ERR_STREAM_DESTROYED') + ) +} + function jsonLine(item: EventStreamItem) { return { schemaVersion: 1, diff --git a/test/commands/event-stream.test.ts b/test/commands/event-stream.test.ts index b23249a3..4d373e94 100644 --- a/test/commands/event-stream.test.ts +++ b/test/commands/event-stream.test.ts @@ -152,6 +152,25 @@ describe('event stream', () => { expect(output).toHaveBeenCalledTimes(1) }) + test('a closed stdout pipe ends the stream without an error', async () => { + const errorListeners = process.stdout.listenerCount('error') + const output = vi.spyOn(process.stdout, 'write').mockImplementation(() => { + setImmediate(() => + process.stdout.emit( + 'error', + Object.assign(new Error('write EPIPE'), {code: 'EPIPE'}), + ), + ) + return false + }) + vi.spyOn(process.stderr, 'write').mockImplementation(() => true) + await environment(() => + EventStream.run(['--owner', 'project', '--project', PROJECT, '--json']), + ) + expect(output).toHaveBeenCalledTimes(1) + expect(process.stdout.listenerCount('error')).toBe(errorListeners) + }) + test('exits with the stream gap code when --on-gap stop sees expired events', async () => { sdk.gap = {missedEvents: 3} vi.spyOn(process.stdout, 'write').mockImplementation(() => true) From 4ec9eb9d0bbfe0387d35bac34d536270e8926c7b Mon Sep 17 00:00:00 2001 From: Rajeh Taher Date: Tue, 22 Sep 2026 14:49:23 +0200 Subject: [PATCH 5/5] fix: stop event streams when stdout cannot drain --- README.md | 3 ++ src/commands/event/stream.ts | 14 +++++- test/commands/event-stream-process.test.ts | 50 ++++++++++++++++++++ test/commands/event-stream.test.ts | 1 + test/fixtures/event-stream-blocked-output.ts | 41 ++++++++++++++++ 5 files changed, 107 insertions(+), 2 deletions(-) create mode 100644 test/commands/event-stream-process.test.ts create mode 100644 test/fixtures/event-stream-blocked-output.ts diff --git a/README.md b/README.md index 90745e92..987d1603 100644 --- a/README.md +++ b/README.md @@ -1088,6 +1088,9 @@ command reconnects with backoff after a dropped connection. If events after `--on-gap stop` to exit with the stream gap code instead. `--max-events` stops after a fixed number of events. +Ctrl-C or SIGTERM stops the stream with exit code 0. If the output pipe is +blocked, stopping discards queued output instead of waiting for the reader. + Use `event stream` to consume events in a script. Use `listen` to forward exact bodies to a local webhook handler. diff --git a/src/commands/event/stream.ts b/src/commands/event/stream.ts index db4dcd42..ce7dea4d 100644 --- a/src/commands/event/stream.ts +++ b/src/commands/event/stream.ts @@ -48,7 +48,11 @@ export default class EventStream extends BaseCommand { const json = flags.json === true const quiet = flags.quiet === true const controller = new AbortController() - const stop = () => controller.abort() + let interrupted = false + const stop = () => { + interrupted = true + controller.abort() + } process.once('SIGINT', stop) process.once('SIGTERM', stop) // A consumer that closes the pipe (for example `| head -n 1`) ends the @@ -115,6 +119,11 @@ export default class EventStream extends BaseCommand { if (!json && !quiet && lastCursor !== undefined) { process.stderr.write(`Resume with --from ${lastCursor}\n`) } + // Aborting the drain wait does not cancel Node's pending stdout write. + // stdout.destroy() also leaves that write alive. After closing the + // iterator and removing listeners, discard the cancelled output so an + // unread pipe cannot keep an interrupted command running indefinitely. + if (interrupted && process.stdout.writableNeedDrain) process.exit(0) } if (gap !== undefined && flags['on-gap'] === 'stop') { throw new CliConditionError( @@ -137,7 +146,8 @@ async function writeStdout(value: string, signal: AbortSignal): Promise { // even when the consumer never drains stdout. await once(process.stdout, 'drain', {signal}) } catch (error) { - if (signal.aborted || isClosedPipe(error)) return + if (isClosedPipe(error)) return + if (signal.aborted) return throw error } } diff --git a/test/commands/event-stream-process.test.ts b/test/commands/event-stream-process.test.ts new file mode 100644 index 00000000..11f2d2b0 --- /dev/null +++ b/test/commands/event-stream-process.test.ts @@ -0,0 +1,50 @@ +import {fork} from 'node:child_process' +import {once} from 'node:events' +import {fileURLToPath} from 'node:url' +import {expect, test} from 'vitest' + +test.each(['SIGINT', 'SIGTERM'] as const)( + '%s exits while a real stdout pipe remains unread', + async (signal) => { + const child = fork( + fileURLToPath( + new URL('../fixtures/event-stream-blocked-output.ts', import.meta.url), + ), + [], + {execArgv: ['--import', 'tsx'], stdio: ['ignore', 'pipe', 'pipe', 'ipc']}, + ) + const deadline = AbortSignal.timeout(5000) + const exited = once(child, 'exit') + let stderr = '' + child.stderr?.setEncoding('utf8').on('data', (value: string) => { + stderr += value + }) + try { + const messages: unknown[] = await once(child, 'message', { + signal: deadline, + }) + expect(messages[0]).toEqual({backpressured: true}) + // Deliberately never read child.stdout, including after the signal. + child.kill(signal) + const result = await Promise.race([ + exited, + new Promise((_, reject) => { + deadline.addEventListener( + 'abort', + () => reject(new Error(`Child did not exit: ${stderr}`)), + {once: true}, + ) + }), + ]) + expect(result).toEqual([0, null]) + expect(stderr).toBe('') + } finally { + if (child.exitCode === null && child.signalCode === null) + child.kill('SIGKILL') + await exited + child.stdout?.destroy() + child.stderr?.destroy() + } + }, + 10000, +) diff --git a/test/commands/event-stream.test.ts b/test/commands/event-stream.test.ts index 4d373e94..26f215fc 100644 --- a/test/commands/event-stream.test.ts +++ b/test/commands/event-stream.test.ts @@ -150,6 +150,7 @@ describe('event stream', () => { EventStream.run(['--owner', 'project', '--project', PROJECT, '--json']), ) expect(output).toHaveBeenCalledTimes(1) + expect(sdk.params[0]?.signal).toMatchObject({aborted: true}) }) test('a closed stdout pipe ends the stream without an error', async () => { diff --git a/test/fixtures/event-stream-blocked-output.ts b/test/fixtures/event-stream-blocked-output.ts new file mode 100644 index 00000000..579159c0 --- /dev/null +++ b/test/fixtures/event-stream-blocked-output.ts @@ -0,0 +1,41 @@ +import {fileURLToPath} from 'node:url' + +import EventStream from '../../src/commands/event/stream.js' + +Object.defineProperty(EventStream.prototype, 'developerSession', { + value: () => + Promise.resolve({ + context: {owner: 'project'}, + client: { + events: { + stream: () => ({ + async *[Symbol.asyncIterator]() { + await Promise.resolve() + setImmediate(() => { + process.send?.({ + backpressured: process.stdout.writableNeedDrain, + }) + }) + yield { + cursor: 'test-cursor', + event: {id: 'test-event', type: 'message.received'}, + webhook: {payload: {text: 'x'.repeat(1024 * 1024)}}, + } + }, + }), + }, + }, + }), +}) + +await EventStream.run( + [ + '--owner', + 'project', + '--project', + '11111111-1111-4111-8111-111111111111', + '--json', + ], + fileURLToPath(new URL('../../', import.meta.url)), +) +process.disconnect?.()