From 6f8384343a816ab31c22c12ecf31d373b95be7a4 Mon Sep 17 00:00:00 2001 From: WolfieLeader Date: Tue, 6 Oct 2026 16:33:56 +0300 Subject: [PATCH 1/2] fix(pg-cloudflare): do not emit an error for the read that end() cancels Since #3747, end() closes the Cloudflare socket while _listen() still waits on a read. workerd rejects that read with "This socket has been closed.", and pg re-emits it as a client 'error' after end() resolved. With no 'error' listener this is an unhandled rejection. Ignore a rejected read once end() has been called. Other read failures still emit 'error'. --- packages/pg-cloudflare/src/index.ts | 12 +++++++++++- packages/pg-esm-test/pg-cloudflare.test.js | 19 +++++++++++++++++++ 2 files changed, 30 insertions(+), 1 deletion(-) diff --git a/packages/pg-cloudflare/src/index.ts b/packages/pg-cloudflare/src/index.ts index 527b433f5..690177fcb 100644 --- a/packages/pg-cloudflare/src/index.ts +++ b/packages/pg-cloudflare/src/index.ts @@ -10,6 +10,7 @@ export class CloudflareSocket extends EventEmitter { private _upgrading = false private _upgraded = false + private _ending = false private _cfSocket: Socket | null = null private _cfWriter: WritableStreamDefaultWriter | null = null private _cfReader: ReadableStreamDefaultReader | null = null @@ -65,7 +66,15 @@ export class CloudflareSocket extends EventEmitter { // eslint-disable-next-line no-constant-condition while (true) { log('awaiting receive from CF socket') - const { done, value } = await this._cfReader!.read() + let result: ReadableStreamReadResult + try { + result = await this._cfReader!.read() + } catch (e) { + // closing the socket in end() rejects the read that is still pending + if (this._ending) break + throw e + } + const { done, value } = result log('CF socket received:', done, value) if (done) { log('done') @@ -110,6 +119,7 @@ export class CloudflareSocket extends EventEmitter { end(data = Buffer.alloc(0), encoding: BufferEncoding = 'utf8', callback: (...args: unknown[]) => void = () => {}) { log('ending CF socket') + this._ending = true this.write(data, encoding, (err) => { const socket = this._cfSocket const closePromise = socket?.close() diff --git a/packages/pg-esm-test/pg-cloudflare.test.js b/packages/pg-esm-test/pg-cloudflare.test.js index 683229158..6f9be18d9 100644 --- a/packages/pg-esm-test/pg-cloudflare.test.js +++ b/packages/pg-esm-test/pg-cloudflare.test.js @@ -55,4 +55,23 @@ describe('pg-cloudflare', () => { socket.end() }) }) + + it('should stop listening without an error when end() closes the socket', async () => { + const socket = new CloudflareSocket() + let rejectRead + socket._cfReader = { + read: () => + new Promise((resolve, reject) => { + rejectRead = reject + }), + } + socket._cfSocket = { + close: async () => rejectRead(new TypeError('This socket has been closed.')), + } + + const listening = socket._listen() + socket.end() + + await listening + }) }) From 572b3b4fc00236fce2423f6f31d9de3fd7dccdce Mon Sep 17 00:00:00 2001 From: WolfieLeader Date: Sat, 17 Oct 2026 19:25:03 +0300 Subject: [PATCH 2/2] fix(pg-cloudflare): also ignore a read that fails after the peer closed the socket --- packages/pg-cloudflare/src/index.ts | 7 ++++-- packages/pg-esm-test/pg-cloudflare.test.js | 28 ++++++++++++++++++++++ 2 files changed, 33 insertions(+), 2 deletions(-) diff --git a/packages/pg-cloudflare/src/index.ts b/packages/pg-cloudflare/src/index.ts index 690177fcb..0b399856b 100644 --- a/packages/pg-cloudflare/src/index.ts +++ b/packages/pg-cloudflare/src/index.ts @@ -11,6 +11,7 @@ export class CloudflareSocket extends EventEmitter { private _upgrading = false private _upgraded = false private _ending = false + private _closed = false private _cfSocket: Socket | null = null private _cfWriter: WritableStreamDefaultWriter | null = null private _cfReader: ReadableStreamDefaultReader | null = null @@ -70,8 +71,9 @@ export class CloudflareSocket extends EventEmitter { try { result = await this._cfReader!.read() } catch (e) { - // closing the socket in end() rejects the read that is still pending - if (this._ending) break + // closing the socket in end() rejects the read that is still pending, + // and after the peer closed it, 'close' has already reported the end + if (this._ending || this._closed) break throw e } const { done, value } = result @@ -166,6 +168,7 @@ export class CloudflareSocket extends EventEmitter { log('CF socket closed') if (this._cfSocket === socket) { this._cfSocket = null + this._closed = true this.emit('close') } } else { diff --git a/packages/pg-esm-test/pg-cloudflare.test.js b/packages/pg-esm-test/pg-cloudflare.test.js index 6f9be18d9..ebcf8eec8 100644 --- a/packages/pg-esm-test/pg-cloudflare.test.js +++ b/packages/pg-esm-test/pg-cloudflare.test.js @@ -74,4 +74,32 @@ describe('pg-cloudflare', () => { await listening }) + + it('should stop listening without an error when the read fails after the socket closed', async () => { + const socket = new CloudflareSocket() + let rejectRead + socket._cfReader = { + read: () => + new Promise((resolve, reject) => { + rejectRead = reject + }), + } + const underlyingSocket = { closed: Promise.resolve() } + socket._cfSocket = underlyingSocket + socket._addClosedHandler() + + const listening = socket._listen() + await underlyingSocket.closed + assert.equal(socket._cfSocket, null) + rejectRead(new TypeError('This socket has been closed.')) + + await listening + }) + + it('should still fail listening when a read fails before end()', async () => { + const socket = new CloudflareSocket() + socket._cfReader = { read: () => Promise.reject(new Error('Network connection lost.')) } + + await assert.rejects(socket._listen(), /Network connection lost/) + }) })