Run tests with github actions

.github/workflows/test.yml

@@ -0,0 +1,47 @@
+name: test
+
+on: [push, pull_request]
+
+jobs:
+  test:
+    name: Test Node v${{ matrix.node }}
+    strategy:
+      matrix:
+        node: ['12', '14', '16', '17']
+    runs-on: ubuntu-latest
+    services:
+      postgres:
+        image: postgres
+        env:
+          POSTGRES_USER: postgres
+          POSTGRES_HOST_AUTH_METHOD: trust
+        ports:
+          - 5433:5432
+        options: >-
+          --health-cmd pg_isready
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+    steps:
+      - uses: actions/checkout@v3
+      - run: |
+          date
+          sudo cp ./tests/pg_hba.conf /etc/postgresql/14/main/pg_hba.conf
+          sudo sed -i 's/.*wal_level.*/wal_level = logical/' /etc/postgresql/14/main/postgresql.conf
+          sudo sed -i 's/.*ssl = .*/ssl = on/' /etc/postgresql/14/main/postgresql.conf
+          openssl req -new -x509 -nodes -days 365 -text -subj "/CN=localhost" -extensions v3_req -config <(cat /etc/ssl/openssl.cnf <(printf "\n[v3_req]\nbasicConstraints=critical,CA:TRUE\nkeyUsage=nonRepudiation,digitalSignature,keyEncipherment\nsubjectAltName=DNS:localhost")) -keyout server.key -out server.crt
+          sudo cp server.key /etc/postgresql/14/main/server.key
+          sudo cp server.crt /etc/postgresql/14/main/server.crt
+          sudo chmod og-rwx /etc/postgresql/14/main/server.key
+          sudo systemctl start postgresql.service
+          pg_isready
+      - uses: denoland/setup-deno@v1
+        with:
+          deno-version: v1.x
+      - uses: actions/setup-node@v3
+        with:
+          node-version: ${{ matrix.node }}
+      - run: npm test
+        env:
+          PGUSER: postgres
+          PGSOCKET: /var/run/postgresql

cjs/src/connection.js

@@ -818,12 +818,14 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose
 
   function CopyInResponse() {
     stream = new Stream.Writable({
+      autoDestroy: true,
       write(chunk, encoding, callback) {
         socket.write(b().d().raw(chunk).end(), callback)
       },
       destroy(error, callback) {
         callback(error)
         socket.write(b().f().str(error + b.N).end())
+        stream = null
       },
       final(callback) {
         socket.write(b().c().end())
@@ -843,6 +845,7 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose
   /* c8 ignore next 3 */
   function CopyBothResponse() {
     stream = new Stream.Duplex({
+      autoDestroy: true,
       read() { socket.resume() },
       /* c8 ignore next 11 */
       write(chunk, encoding, callback) {
@@ -851,6 +854,7 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose
       destroy(error, callback) {
         callback(error)
         socket.write(b().f().str(error + b.N).end())
+        stream = null
       },
       final(callback) {
         socket.write(b().c().end())

cjs/src/subscribe.js

@@ -66,7 +66,7 @@ module.exports = Subscribe;function Subscribe(postgres, options) {
 
     stream.on('data', data)
     stream.on('error', (error) => {
-      console.error('Logical Replication Error - Reconnecting', error)
+      console.error('Logical Replication Error - Reconnecting', error) // eslint-disable-line
       sql.end()
     })
 

cjs/tests/index.js

@@ -1,5 +1,3 @@
-/* eslint no-console: 0 */
-
 const { exec } = require('./bootstrap.js')
 
 const { t, nt, ot } = require('./test.js') // eslint-disable-line
@@ -666,18 +664,19 @@ t('listen and notify with upper case', async() => {
 
 t('listen reconnects', { timeout: 2 }, async() => {
   const sql = postgres(options)
-      , xs = []
+      , resolvers = {}
+      , a = new Promise(r => resolvers.a = r)
+      , b = new Promise(r => resolvers.b = r)
 
-  const { state: { pid } } = await sql.listen('test', x => xs.push(x))
-  await delay(200)
+  const { state: { pid } } = await sql.listen('test', x => x in resolvers && resolvers[x]())
   await sql.notify('test', 'a')
-  await sql`select pg_terminate_backend(${ pid }::int)`
-  await delay(200)
+  await a
+  await sql`select pg_terminate_backend(${ pid })`
+  await delay(50)
   await sql.notify('test', 'b')
-  await delay(200)
+  await b
   sql.end()
-
-  return ['ab', xs.join('')]
+  return [true, true]
 })
 
 
@@ -687,7 +686,7 @@ t('listen reconnects after connection error', { timeout: 3 }, async() => {
 
   const { state: { pid } } = await sql.listen('test', x => xs.push(x))
   await sql.notify('test', 'a')
-  await sql`select pg_terminate_backend(${ pid }::int)`
+  await sql`select pg_terminate_backend(${ pid })`
   await delay(1000)
 
   await sql.notify('test', 'b')
@@ -704,7 +703,7 @@ t('listen result reports correct connection state after reconnection', async() =
   const result = await sql.listen('test', x => xs.push(x))
   const initialPid = result.state.pid
   await sql.notify('test', 'a')
-  await sql`select pg_terminate_backend(${ initialPid }::int)`
+  await sql`select pg_terminate_backend(${ initialPid })`
   await delay(50)
   sql.end()
 
@@ -852,7 +851,7 @@ t('Connection errors are caught using begin()', {
 }, async() => {
   let error
   try {
-    const sql = postgres({ host: 'wat', port: 1337 })
+    const sql = postgres({ host: 'localhost', port: 1 })
 
     await sql.begin(async(sql) => {
       await sql`insert into test (label, value) values (${1}, ${2})`
@@ -863,8 +862,8 @@ t('Connection errors are caught using begin()', {
 
   return [
     true,
-    error.code === 'ENOTFOUND' ||
-    error.message === 'failed to lookup address information: nodename nor servname provided, or not known'
+    error.code === 'ECONNREFUSED' ||
+    error.message === 'Connection refused (os error 61)'
   ]
 })
 
@@ -1016,8 +1015,8 @@ t('throws correct error when authentication fails', async() => {
 
 t('notice works', async() => {
   let notice
-  const log = console.log
-  console.log = function(x) {
+  const log = console.log // eslint-disable-line
+  console.log = function(x) { // eslint-disable-line
     notice = x
   }
 
@@ -1026,7 +1025,7 @@ t('notice works', async() => {
   await sql`create table if not exists users()`
   await sql`create table if not exists users()`
 
-  console.log = log
+  console.log = log // eslint-disable-line
 
   return ['NOTICE', notice.severity]
 })
@@ -1252,7 +1251,7 @@ t('Transform columns from', async() => {
 t('Unix socket', async() => {
   const sql = postgres({
     ...options,
-    host: '/tmp'
+    host: process.env.PGSOCKET || '/tmp' // eslint-disable-line
   })
 
   return [1, (await sql`select 1 as x`)[0].x]
@@ -1378,7 +1377,7 @@ t('requests works after single connect_timeout', async() => {
   const sql = postgres({
     ...options,
     ...login_scram,
-    connect_timeout: { valueOf() { return first ? (first = false, 0.001) : 1 } }
+    connect_timeout: { valueOf() { return first ? (first = false, 0.0001) : 1 } }
   })
 
   return [
@@ -1538,19 +1537,22 @@ t('Multiple hosts', {
       , sql = postgres('postgres://localhost:5432,localhost:5433', { idle_timeout, max: 1 })
       , result = []
 
+  const id1 = (await s1`select system_identifier as x from pg_control_system()`)[0].x
+  const id2 = (await s2`select system_identifier as x from pg_control_system()`)[0].x
+
   const x1 = await sql`select 1`
-  result.push((await sql`select setting as x from pg_settings where name = 'port'`)[0].x)
+  result.push((await sql`select system_identifier as x from pg_control_system()`)[0].x)
   await s1`select pg_terminate_backend(${ x1.state.pid }::int)`
   await delay(100)
 
   const x2 = await sql`select 1`
-  result.push((await sql`select setting as x from pg_settings where name = 'port'`)[0].x)
+  result.push((await sql`select system_identifier as x from pg_control_system()`)[0].x)
   await s2`select pg_terminate_backend(${ x2.state.pid }::int)`
   await delay(100)
 
-  result.push((await sql`select setting as x from pg_settings where name = 'port'`)[0].x)
+  result.push((await sql`select system_identifier as x from pg_control_system()`)[0].x)
 
-  return ['5432,5433,5432', result.join(',')]
+  return [[id1, id2, id1].join(','), result.join(',')]
 })
 
 t('Escaping supports schemas and tables', async() => {
@@ -1762,9 +1764,9 @@ t('Cancel running query works', async() => {
 
 t('Cancel piped query works', { timeout: 1 }, async() => {
   await sql`select 1`
-  const last = sql`select pg_sleep(0.2)`.execute()
+  const last = sql`select pg_sleep(0.3)`.execute()
   const query = sql`select pg_sleep(2) as dig`
-  setTimeout(() => query.cancel(), 100)
+  setTimeout(() => query.cancel(), 10)
   const error = await query.catch(x => x)
   await last
   return ['57014', error.code]
@@ -1773,7 +1775,7 @@ t('Cancel piped query works', { timeout: 1 }, async() => {
 t('Cancel queued query works', async() => {
   const tx = sql.begin(sql => sql`select pg_sleep(0.2) as hej, 'hejsa'`)
   const query = sql`select pg_sleep(2) as nej`
-  setTimeout(() => query.cancel(), 50)
+  setTimeout(() => query.cancel(), 100)
   const error = await query.catch(x => x)
   await tx
   return ['57014', error.code]
@@ -1942,7 +1944,7 @@ t('Prevent premature end of connection in transaction', async() => {
   ]
 })
 
-t('Ensure reconnect after max_lifetime with transactions', { timeout: 5000 }, async() => {
+t('Ensure reconnect after max_lifetime with transactions', { timeout: 5 }, async() => {
   const sql = postgres({
     max_lifetime: 0.01,
     idle_timeout,

cjs/tests/pg_hba.conf

@@ -0,0 +1,5 @@
+local   all                 all                                     trust
+host    all                 postgres                samehost        trust
+host    postgres_js_test    postgres_js_test        samehost        trust
+host    postgres_js_test    postgres_js_test_md5    samehost        md5
+host    postgres_js_test    postgres_js_test_scram  samehost        scram-sha-256

deno/polyfills.js

@@ -28,8 +28,8 @@ export const net = {
     const socket = {
       error,
       success,
+      readyState: 'closed',
       connect: (port, hostname) => {
-        socket.closed = false
         socket.raw = null
         typeof port === 'string'
           ? Deno.connect({ transport: 'unix', path: socket.path = port }).then(success, error)
@@ -79,6 +79,7 @@ export const net = {
 
     async function success(raw) {
       const encrypted = socket.encrypted
+      socket.readyState = 'open'
       socket.raw = raw
       socket.encrypted
         ? call(socket.events.secureConnect)
@@ -88,7 +89,7 @@ export const net = {
       let result
 
       try {
-        while ((result = !socket.closed && await raw.read(b))) {
+        while ((result = socket.readyState === 'open' && await raw.read(b))) {
           call(socket.events.data, Buffer.from(b.subarray(0, result)))
           if (!encrypted && socket.break && (socket.break = false, b[0] === 83))
             return socket.break = false
@@ -115,11 +116,11 @@ export const net = {
 
     function closed() {
       socket.break = socket.encrypted = false
-      if (socket.closed)
+      if (socket.readyState !== 'open')
         return
 
       call(socket.events.close)
-      socket.closed = true
+      socket.readyState = 'closed'
     }
 
     function error(err) {

deno/src/connection.js

@@ -821,12 +821,14 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose
 
   function CopyInResponse() {
     stream = new Stream.Writable({
+      autoDestroy: true,
       write(chunk, encoding, callback) {
         socket.write(b().d().raw(chunk).end(), callback)
       },
       destroy(error, callback) {
         callback(error)
         socket.write(b().f().str(error + b.N).end())
+        stream = null
       },
       final(callback) {
         socket.write(b().c().end())
@@ -846,6 +848,7 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose
   /* c8 ignore next 3 */
   function CopyBothResponse() {
     stream = new Stream.Duplex({
+      autoDestroy: true,
       read() { socket.resume() },
       /* c8 ignore next 11 */
       write(chunk, encoding, callback) {
@@ -854,6 +857,7 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose
       destroy(error, callback) {
         callback(error)
         socket.write(b().f().str(error + b.N).end())
+        stream = null
       },
       final(callback) {
         socket.write(b().c().end())

deno/src/subscribe.js

@@ -67,7 +67,7 @@ export default function Subscribe(postgres, options) {
 
     stream.on('data', data)
     stream.on('error', (error) => {
-      console.error('Logical Replication Error - Reconnecting', error)
+      console.error('Logical Replication Error - Reconnecting', error) // eslint-disable-line
       sql.end()
     })