feat: add log and more tests

Author: panga

packages/pg-pool/index.js

@@ -116,6 +116,10 @@ class Pool extends EventEmitter {
     return this._clients.length > this.options.min
   }
 
+  _hasActiveTransaction(client) {
+    return client && (client._txStatus === 'T' || client._txStatus === 'E')
+  }
+
   _pulseQueue() {
     this.log('pulse queue')
     if (this.ended) {
@@ -364,13 +368,15 @@ class Pool extends EventEmitter {
       this.ending ||
       !client._queryable ||
       client._ending ||
-      client._txStatus === 'T' ||
-      client._txStatus === 'E' ||
+      this._hasActiveTransaction(client) ||
       client._poolUseCount >= this.options.maxUses
     ) {
       if (client._poolUseCount >= this.options.maxUses) {
         this.log('remove expended client')
       }
+      if (this._hasActiveTransaction(client)) {
+        this.log('remove client with leaked transaction')
+      }
 
       return this._remove(client, this._pulseQueue.bind(this))
     }

packages/pg-pool/test/poison-pool.js packages/pg-pool/test/leaked-pool.jspackages/pg-pool/test/poison-pool.js renamed to packages/pg-pool/test/leaked-pool.js

@@ -5,20 +5,28 @@ const describe = require('mocha').describe
 const it = require('mocha').it
 const Pool = require('..')
 
-describe('poison connection pool defense (_txStatus check)', function () {
+describe('leaked connection pool guard', function () {
   it('removes a client with an open transaction on release', async function () {
-    const pool = new Pool({ max: 1 })
+    const logMessages = []
+    const pool = new Pool({
+      max: 1,
+      log: (msg) => logMessages.push(msg),
+    })
     const client = await pool.connect()
     await client.query('BEGIN')
     expect(client._txStatus).to.be('T')
 
     client.release()
     expect(pool.totalCount).to.be(0)
     expect(pool.idleCount).to.be(0)
+    expect(logMessages).to.contain('remove client with leaked transaction')
 
-    // pool should still work by creating a fresh connection
+    // pool recovers by creating a fresh connection
     const { rows } = await pool.query('SELECT 1 as num')
     expect(rows[0].num).to.be(1)
+    expect(pool.totalCount).to.be(1)
+    expect(pool.idleCount).to.be(1)
+
     await pool.end()
   })
 
@@ -45,9 +53,12 @@ describe('poison connection pool defense (_txStatus check)', function () {
     expect(pool.totalCount).to.be(0)
     expect(pool.idleCount).to.be(0)
 
-    // pool should still work
+    // pool recovers by creating a fresh connection
     const { rows } = await pool.query('SELECT 1 as num')
     expect(rows[0].num).to.be(1)
+    expect(pool.totalCount).to.be(1)
+    expect(pool.idleCount).to.be(1)
+
     await pool.end()
   })
 
@@ -57,7 +68,7 @@ describe('poison connection pool defense (_txStatus check)', function () {
     const clientB = await pool.connect()
     const clientC = await pool.connect()
 
-    // Client A: open transaction (poisoned)
+    // Client A: open transaction (leaked)
     await clientA.query('BEGIN')
     expect(clientA._txStatus).to.be('T')
 
@@ -79,4 +90,50 @@ describe('poison connection pool defense (_txStatus check)', function () {
     expect(pool.idleCount).to.be(2)
     await pool.end()
   })
+
+  describe('pool.query', function () {
+    it('removes a client after pool.query leaks transaction via BEGIN', async function () {
+      const logMessages = []
+      const pool = new Pool({ max: 1, log: (msg) => logMessages.push(msg) })
+
+      await pool.query('BEGIN')
+
+      // Client auto-released with txStatus='T', should be removed
+      expect(pool.totalCount).to.be(0)
+      expect(pool.idleCount).to.be(0)
+      expect(logMessages).to.contain('remove client with leaked transaction')
+
+      // Verify pool recovers
+      const { rows } = await pool.query('SELECT 1 as num')
+      expect(rows[0].num).to.be(1)
+      expect(pool.totalCount).to.be(1)
+      expect(pool.idleCount).to.be(1)
+
+      await pool.end()
+    })
+
+    it('removes a client after pool.query in failed transaction state', async function () {
+      const pool = new Pool({ max: 1 })
+
+      await pool.query('BEGIN')
+
+      try {
+        await pool.query('SELECT invalid_column FROM nonexistent_table')
+      } catch (e) {
+        // Expected error
+      }
+
+      // Client with txStatus='E' should be removed
+      expect(pool.totalCount).to.be(0)
+      expect(pool.idleCount).to.be(0)
+
+      // Pool recovers
+      const { rows } = await pool.query('SELECT 1 as num')
+      expect(rows[0].num).to.be(1)
+      expect(pool.totalCount).to.be(1)
+      expect(pool.idleCount).to.be(1)
+
+      await pool.end()
+    })
+  })
 })

packages/pg/test/integration/client/txstatus-tests.js

@@ -0,0 +1,76 @@
+'use strict'
+const helper = require('./test-helper')
+const suite = new helper.Suite()
+const pg = helper.pg
+const assert = require('assert')
+
+suite.test('txStatus tracking', function (done) {
+  const client = new pg.Client()
+  client.connect(
+    assert.success(function () {
+      // Run a simple query to initialize txStatus
+      client.query(
+        'SELECT 1',
+        assert.success(function () {
+          // Test 1: Initial state after query (should be idle)
+          assert.equal(client._txStatus, 'I', 'should start in idle state')
+
+          // Test 2: BEGIN transaction
+          client.query(
+            'BEGIN',
+            assert.success(function () {
+              assert.equal(client._txStatus, 'T', 'should be in transaction state')
+
+              // Test 3: COMMIT
+              client.query(
+                'COMMIT',
+                assert.success(function () {
+                  assert.equal(client._txStatus, 'I', 'should return to idle after commit')
+
+                  client.end(done)
+                })
+              )
+            })
+          )
+        })
+      )
+    })
+  )
+})
+
+suite.test('txStatus error state', function (done) {
+  const client = new pg.Client()
+  client.connect(
+    assert.success(function () {
+      // Run a simple query to initialize txStatus
+      client.query(
+        'SELECT 1',
+        assert.success(function () {
+          client.query(
+            'BEGIN',
+            assert.success(function () {
+              // Execute invalid SQL to trigger error state
+              client.query('INVALID SQL SYNTAX', function (err) {
+                assert(err, 'should receive error from invalid query')
+
+                // Use setImmediate to allow ReadyForQuery message to be processed
+                setImmediate(function () {
+                  assert.equal(client._txStatus, 'E', 'should be in error state')
+
+                  // Rollback to recover
+                  client.query(
+                    'ROLLBACK',
+                    assert.success(function () {
+                      assert.equal(client._txStatus, 'I', 'should return to idle after rollback from error')
+                      client.end(done)
+                    })
+                  )
+                })
+              })
+            })
+          )
+        })
+      )
+    })
+  )
+})