fix(pg-cloudflare): safely end already-closed sockets

packages/pg-cloudflare/package.json

@@ -6,6 +6,8 @@
   "types": "dist/index.d.ts",
   "license": "MIT",
   "devDependencies": {
+    "@types/mocha": "^10.0.10",
+    "mocha": "^11.7.5",
     "ts-node": "^8.5.4",
     "typescript": "^6.0.3"
   },
@@ -23,7 +25,8 @@
     "build": "tsc",
     "build:watch": "tsc --watch",
     "prepublish": "yarn build",
-    "test": "echo e2e test in pg package"
+    "pretest": "yarn build",
+    "test": "mocha dist/**/*.test.js"
   },
   "repository": {
     "type": "git",

packages/pg-cloudflare/src/index.test.ts

@@ -0,0 +1,62 @@
+import assert from 'assert'
+
+import { CloudflareSocket } from './index'
+
+function deferred() {
+  let resolve!: () => void
+  const promise = new Promise<void>((done) => {
+    resolve = done
+  })
+  return { promise, resolve }
+}
+
+describe('CloudflareSocket', () => {
+  it('closes an open socket and invokes the end callback', () => {
+    const socket = new CloudflareSocket(false)
+    let closeCalls = 0
+    Object.assign(socket, {
+      _cfSocket: {
+        close: () => closeCalls++,
+      },
+    })
+
+    let callbackCalls = 0
+    socket.end(Buffer.alloc(0), 'utf8', (err) => {
+      callbackCalls++
+      assert.equal(err, undefined)
+    })
+
+    assert.equal(closeCalls, 1)
+    assert.equal(callbackCalls, 1)
+  })
+
+  it('invokes the end callback after the closed handler clears the socket', async () => {
+    const socket = new CloudflareSocket(false)
+    const closed = deferred()
+    let closeCalls = 0
+    Object.assign(socket, {
+      _cfSocket: {
+        closed: closed.promise,
+        close: () => closeCalls++,
+      },
+    })
+
+    let closeEvents = 0
+    socket.on('close', () => closeEvents++)
+    socket._addClosedHandler()
+    closed.resolve()
+    await closed.promise
+
+    let callbackCalls = 0
+    assert.doesNotThrow(() => {
+      socket.end(Buffer.alloc(0), 'utf8', (err) => {
+        callbackCalls++
+        assert.equal(err, undefined)
+      })
+    })
+
+    assert.equal(closeCalls, 0)
+    assert.equal(closeEvents, 1)
+    assert.equal(callbackCalls, 1)
+  })
+})

packages/pg-cloudflare/src/index.ts

@@ -107,7 +107,7 @@ export class CloudflareSocket extends EventEmitter {
   end(data = Buffer.alloc(0), encoding: BufferEncoding = 'utf8', callback: (...args: unknown[]) => void = () => {}) {
     log('ending CF socket')
     this.write(data, encoding, (err) => {
-      this._cfSocket!.close()
+      this._cfSocket?.close()
       if (callback) callback(err)
     })
     return this

packages/pg-cloudflare/tsconfig.json

@@ -18,7 +18,7 @@
         "./src/types/*"
       ]
     },
-    "types": ["node"]
+    "types": ["node", "mocha"]
   },
   "include": [
     "src/**/*"