From 846a29896e0e52a368a8f1906bbdee39ce55b38a Mon Sep 17 00:00:00 2001 From: Dias Date: Tue, 8 Sep 2026 09:53:11 -0700 Subject: [PATCH] Bound reconnects after clean startup closes --- src/connection.js | 23 ++++-- tests/connection.js | 186 ++++++++++++++++++++++++++++++++++++++++++++ tests/index.js | 11 +++ 3 files changed, 213 insertions(+), 7 deletions(-) create mode 100644 tests/connection.js diff --git a/src/connection.js b/src/connection.js index 10ab1bb..8d631ed 100644 --- a/src/connection.js +++ b/src/connection.js @@ -87,6 +87,7 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose , statementId = Math.random().toString(36).slice(2) , statementCount = 1 , closedTime = 0 + , closeRunStart = 0 , remaining = 0 , hostIndex = 0 , retries = 0 @@ -447,13 +448,21 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose socket.removeAllListeners() socket = null - if (initial) - return reconnect() - - !hadError && (query || sent.length) && error(Errors.connection('CONNECTION_CLOSED', options, socket)) closedTime = performance.now() - hadError && options.shared.retries++ + if (hadError || initial) + options.shared.retries++ delay = (typeof backoff === 'function' ? backoff(options.shared.retries) : backoff) * 1000 + + if (initial) { + closeRunStart || (closeRunStart = closedTime) + // Do not schedule a retry beyond the connection's clean-close budget. + if (closedTime + delay <= closeRunStart + (options.connect_timeout || 30) * 1000) + return reconnect() + errored(Errors.connection('CONNECTION_CLOSED', options, socket)) + } + + closeRunStart = 0 + !hadError && (query || sent.length) && error(Errors.connection('CONNECTION_CLOSED', options, socket)) onclose(connection, Errors.connection('CONNECTION_CLOSED', options, socket)) } @@ -560,12 +569,12 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose } if (needsTypes) { - initial.reserve && (initial = null) + initial.reserve && (options.shared.retries = retries = closeRunStart = 0, initial = null) return fetchArrayTypes() } initial && !initial.reserve && execute(initial) - options.shared.retries = retries = 0 + options.shared.retries = retries = closeRunStart = 0 initial = null return } diff --git a/tests/connection.js b/tests/connection.js new file mode 100644 index 0000000..633d4be --- /dev/null +++ b/tests/connection.js @@ -0,0 +1,186 @@ +import assert from 'assert' +import { createServer } from 'net' +import postgres from '../src/index.js' + +const delay = ms => new Promise(resolve => setTimeout(resolve, ms)) + , ready = message('Z', [73]) + +export function cleanClose() { + let attempts = 0 + , closes = 0 + const retries = [] + + return peer(() => (++attempts, 'close'), { + connect_timeout: 0.15, + backoff: retry => (retries.push(retry), 0.025), + onclose: () => closes++ + }, async sql => { + const error = await sql`select 1`.catch(error => error) + assert.strictEqual(error.code, 'CONNECTION_CLOSED') + assert(attempts > 1 && attempts < 15) + assert.strictEqual(closes, 1) + assert.deepStrictEqual(retries, retries.map((_, i) => i + 1)) + const settled = attempts + await delay(75) + assert.strictEqual(attempts, settled) + }) +} + +export function queuedClose() { + return peer(() => 'close', { + connect_timeout: 0.1, + backoff: 0.025 + }, async sql => { + const errors = await Promise.all([ + sql`select 1`.catch(error => error.code), + sql`select 2`.catch(error => error.code), + sql.reserve().catch(error => error.code) + ]) + assert.deepStrictEqual(errors, ['CONNECTION_CLOSED', 'CONNECTION_CLOSED', 'CONNECTION_CLOSED']) + }) +} + +export function closeBackoff() { + let attempts = 0 + return peer(() => (++attempts, 'close'), { + connect_timeout: 0.1, + backoff: 2 + }, async sql => { + const start = Date.now() + assert.strictEqual(await sql`select 1`.catch(error => error.code), 'CONNECTION_CLOSED') + assert(Date.now() - start < 500) + assert.strictEqual(attempts, 1) + }) +} + +export function closeRecovery() { + let attempts = 0 + return peer(() => ++attempts <= 6 ? 'close' : 'ready', { + connect_timeout: 0.5, + backoff: 0.025 + }, async sql => { + assert.strictEqual((await sql`select 1`).command, 'SELECT') + assert.strictEqual(attempts, 7) + }) +} + +export function closeReset() { + let attempts = 0 + return peer(() => ++attempts % 3 ? 'close' : 'ready', { + connect_timeout: 0.15, + backoff: 0.025 + }, async sql => { + for (let i = 0; i < 3; i++) { + assert.strictEqual((await sql`select 1`).command, 'SELECT') + await sql.close() + await delay(175) + } + assert.strictEqual(attempts, 9) + }) +} + +export function reserveCloseReset() { + let attempts = 0 + const retries = [] + return peer(() => ++attempts % 3 ? 'close' : 'ready', { + connect_timeout: 0.15, + backoff: retry => (retries.push(retry), 0.025) + }, async sql => { + for (let i = 0; i < 3; i++) { + const reserved = await sql.reserve() + assert.strictEqual((await reserved`select 1`).command, 'SELECT') + reserved.release() + await sql.close() + await delay(175) + } + assert.strictEqual(attempts, 9) + assert.deepStrictEqual(retries.filter(retry => retry > 0), [1, 2, 1, 2, 1, 2]) + }) +} + +export function closeErrorReset() { + let attempts = 0 + return peer(() => ++attempts === 2 ? 'error' : attempts < 5 ? 'close' : 'ready', { + connect_timeout: 0.15, + backoff: 0.025 + }, async sql => { + assert.strictEqual(await sql`select 1`.catch(error => error.code), '53300') + await delay(175) + assert.strictEqual((await sql`select 1`).command, 'SELECT') + assert.strictEqual(attempts, 5) + }) +} + +async function peer(accept, options, run) { + const sockets = new Set() + , server = createServer(socket => { + sockets.add(socket) + socket.on('close', () => sockets.delete(socket)) + const action = accept() + let incoming = Buffer.alloc(0) + , startup = true + socket.on('data', data => { + const responses = [] + incoming = Buffer.concat([incoming, data]) + while (incoming.length >= (startup ? 4 : 5)) { + const length = startup ? incoming.readUInt32BE(0) : incoming.readUInt32BE(1) + 1 + if (incoming.length < length) + break + const type = incoming[0] + incoming = incoming.subarray(length) + if (startup) { + startup = false + if (action === 'close') + return socket.end() + if (action === 'error') + return socket.end(message('E', Buffer.from('SFATAL\0C53300\0Mtoo many connections\0\0'))) + responses.push(message('R', [0, 0, 0, 0]), ready) + } else if (type === 80) { // Parse + responses.push(message('1')) + } else if (type === 68) { // Describe + responses.push(message('t', [0, 0]), message('n')) + } else if (type === 66) { // Bind + responses.push(message('2')) + } else if (type === 69 || type === 81) { // Execute or Query + responses.push(message('C', Buffer.from('SELECT 0\0'))) + type === 81 && responses.push(ready) + } else if (type === 83) { // Sync + responses.push(ready) + } else if (type === 88) { // Terminate + socket.end() + } + } + responses.length && socket.write(Buffer.concat(responses)) + }) + }) + + await new Promise(resolve => server.listen(0, '127.0.0.1', resolve)) + const sql = postgres({ + host: '127.0.0.1', + port: server.address().port, + user: 'test', + database: 'test', + ssl: false, + max: 1, + prepare: false, + ...options + }) + const timeout = setTimeout(() => sql.end({ timeout: 0 }), 2000) + try { + await run(sql) + return [true, true] + } finally { + clearTimeout(timeout) + await sql.end({ timeout: 0 }) + sockets.forEach(socket => socket.destroy()) + await new Promise(resolve => server.close(resolve)) + } +} + +function message(type, data = []) { + const payload = Buffer.from(data) + , header = Buffer.alloc(5) + header[0] = type.charCodeAt(0) + header.writeUInt32BE(payload.length + 4, 1) + return Buffer.concat([header, payload]) +} diff --git a/tests/index.js b/tests/index.js index 845c6ce..aee86f1 100644 --- a/tests/index.js +++ b/tests/index.js @@ -6,6 +6,9 @@ import fs from 'fs' import crypto from 'crypto' import postgres from '../src/index.js' +import { + cleanClose, queuedClose, closeBackoff, closeRecovery, closeReset, reserveCloseReset, closeErrorReset +} from './connection.js' const delay = ms => new Promise(r => setTimeout(r, ms)) const rel = x => new URL(x, import.meta.url) @@ -1650,6 +1653,14 @@ t('Query and parameters are enumerable if debug is set', async() => { ] }) +t('Clean closes reject the initial query with backoff', cleanClose) +t('Clean closes settle queued queries and reserves', queuedClose) +t('Clean closes reject before a backoff beyond connect_timeout', closeBackoff) +t('Initial query recovers after more than five clean closes', closeRecovery) +t('Clean close deadline resets after a successful query', closeReset) +t('Clean close deadline resets after reserve with fetch_types', reserveCloseReset) +t('Clean close deadline resets after a startup error', closeErrorReset) + t('connect_timeout', { timeout: 20 }, async() => { const connect_timeout = 0.2 const server = net.createServer()