diff --git a/.changeset/handle-shape-stream-start-errors.md b/.changeset/handle-shape-stream-start-errors.md new file mode 100644 index 0000000000..e40be21e25 --- /dev/null +++ b/.changeset/handle-shape-stream-start-errors.md @@ -0,0 +1,6 @@ +--- +"@electric-sql/client": patch +--- + +Prevent subscription startup failures from also surfacing as unhandled promise +rejections after they are delivered to the subscriber error callback. diff --git a/packages/typescript-client/src/client.ts b/packages/typescript-client/src/client.ts index 4e2504c9e9..60d48c9b38 100644 --- a/packages/typescript-client/src/client.ts +++ b/packages/typescript-client/src/client.ts @@ -1825,7 +1825,12 @@ export class ShapeStream = Row> const subscriptionId = {} this.#subscribers.set(subscriptionId, [callback, onError]) - if (!this.#started) this.#start() + if (!this.#started) { + this.#start().catch(() => { + // Errors from #start are handled internally via onError. + // This catch prevents unhandled promise rejection in Node/Bun. + }) + } return () => { this.#subscribers.delete(subscriptionId) diff --git a/packages/typescript-client/test/stream.test.ts b/packages/typescript-client/test/stream.test.ts index ceaeafa8ba..84e76782ee 100644 --- a/packages/typescript-client/test/stream.test.ts +++ b/packages/typescript-client/test/stream.test.ts @@ -1,6 +1,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import { ShapeStream, + FetchError, isChangeMessage, isControlMessage, Message, @@ -25,6 +26,36 @@ describe(`ShapeStream`, () => { afterEach(() => aborter.abort()) + it(`does not create an unhandled rejection when a subscriber handles an error`, async () => { + const unhandledRejections: unknown[] = [] + const onUnhandledRejection = (reason: unknown) => { + unhandledRejections.push(reason) + } + process.on(`unhandledRejection`, onUnhandledRejection) + + try { + const stream = new ShapeStream({ + url: shapeUrl, + params: { table: `test` }, + signal: aborter.signal, + fetchClient: async () => + new Response(undefined, { + status: 401, + }), + }) + + const subscriberError = new Promise((resolve) => { + stream.subscribe(() => {}, resolve) + }) + + await expect(subscriberError).resolves.toBeInstanceOf(FetchError) + await resolveInMacrotask(undefined) + expect(unhandledRejections).toEqual([]) + } finally { + process.off(`unhandledRejection`, onUnhandledRejection) + } + }) + it(`requestSnapshot waits for snapshot messages to be published to subscribers before resolving`, async () => { const snapshotRow = { key: `test-1`,