diff --git a/packages/pg-cloudflare/src/index.ts b/packages/pg-cloudflare/src/index.ts index 527b433f5..98a351b8f 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 @@ -110,6 +111,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() @@ -120,7 +122,14 @@ export class CloudflareSocket extends EventEmitter { this.emit('close') } }) - .catch((e) => this.emit('error', e)) + .catch(() => { + // The socket was being closed anyway; still surface 'close' so + // consumers waiting on it (like Connection.end()) resolve. + if (this._cfSocket === socket) { + this._cfSocket = null + this.emit('close') + } + }) if (callback) callback(err) }) return this @@ -150,20 +159,31 @@ export class CloudflareSocket extends EventEmitter { _addClosedHandler() { const socket = this._cfSocket! - socket.closed - .then(() => { - if (!this._upgrading) { - log('CF socket closed') - if (this._cfSocket === socket) { - this._cfSocket = null - this.emit('close') - } - } else { - this._upgrading = false - this._upgraded = true + const handleClosed = () => { + if (!this._upgrading) { + log('CF socket closed') + if (this._cfSocket === socket) { + this._cfSocket = null + this.emit('close') } - }) - .catch((e) => this.emit('error', e)) + } else { + this._upgrading = false + this._upgraded = true + } + } + socket.closed.then( + () => handleClosed(), + (e) => { + // workerd can reject `closed` even for a normal shutdown + // (cloudflare/workerd#6479). Match net.Socket and still emit 'close' + // so consumers waiting on it resolve. A rejection after end() is the + // expected shutdown path, so the spurious 'error' is skipped then. + if (!this._ending) { + this.emit('error', e) + } + handleClosed() + } + ) } } diff --git a/packages/pg-esm-test/pg-cloudflare.test.js b/packages/pg-esm-test/pg-cloudflare.test.js index 683229158..2ebb93d45 100644 --- a/packages/pg-esm-test/pg-cloudflare.test.js +++ b/packages/pg-esm-test/pg-cloudflare.test.js @@ -38,6 +38,53 @@ describe('pg-cloudflare', () => { await promise }) + it('should emit error then close when the underlying closed promise rejects', async () => { + const socket = new CloudflareSocket() + const cause = new Error('socket died') + socket._cfSocket = { closed: Promise.reject(cause) } + socket._addClosedHandler() + + const order = [] + const seen = await new Promise((resolve) => { + socket.once('error', (error) => { + order.push('error') + resolve(error) + }) + socket.once('close', () => order.push('close')) + }) + + await new Promise((resolve) => setImmediate(resolve)) + assert.equal(seen, cause) + assert.deepEqual(order, ['error', 'close']) + }) + + it('should emit close without error when the closed promise rejects after end()', async () => { + const socket = new CloudflareSocket() + let rejectClosed + socket._cfSocket = { + closed: new Promise((_, reject) => (rejectClosed = reject)), + close: async () => {}, + } + socket._cfWriter = { write: () => Promise.resolve() } + socket._addClosedHandler() + + let errorEmitted = false + socket.on('error', () => (errorEmitted = true)) + const closed = new Promise((resolve, reject) => { + const timer = setTimeout(() => reject(new Error('close event was not emitted')), 100) + socket.once('close', () => { + clearTimeout(timer) + resolve() + }) + }) + + socket.end() + rejectClosed(undefined) + await closed + await new Promise((resolve) => setImmediate(resolve)) + assert.equal(errorEmitted, false) + }) + it('should emit close when ending a socket whose closed promise never settles', async () => { const socket = new CloudflareSocket() socket._cfSocket = {