From 0ecb7e726035f3565e5844d3799bfdee76c52303 Mon Sep 17 00:00:00 2001 From: Fabio Angius Date: Thu, 16 Jul 2026 07:37:12 -0700 Subject: [PATCH] =?UTF-8?q?feat:=20reject=5Fthrottle=20=E2=80=94=20pace=20?= =?UTF-8?q?connection=20opens=20after=20establishment=20failures?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Opt-in (default off; off is behaviourally identical to today). When a connection fails to ESTABLISH, the pool otherwise retries every connection immediately and in lockstep — stampeding a server that is already rejecting connections. With reject_throttle enabled the pool keeps a single failed connection as the lone prober (retrying with backoff), blocks the rest, and on each prober handshake success admits an exponentially growing batch of waiters (slow-start, capped at `max`), disengaging once the backlog drains. This complements the existing per-connection `backoff`: backoff only delays each retry, and since its counter is pool-shared every connection waits the same delay and then fires together — a delayed herd. reject_throttle instead keeps exactly one attempt in flight during an outage. It governs only the clean-close establishment-reconnect path (`if (initial) return reconnect()`); error-close (RST) and connect-timeout still reject as before. Tests in tests/reject_throttle.js (21 cases): option plumbing, throttle-vs-herd, single prober, ramp collapse, recovery and disengagement (including large backlogs and max:1), drop-path intact via max_lifetime, and boundaries (RST reject, connect_timeout, off == vanilla). --- package.json | 2 +- src/connection.js | 65 +++++++- src/index.js | 65 +++++++- tests/reject_throttle.js | 329 +++++++++++++++++++++++++++++++++++++++ 4 files changed, 450 insertions(+), 11 deletions(-) create mode 100644 tests/reject_throttle.js diff --git a/package.json b/package.json index c3b76a1a..a05d1e70 100644 --- a/package.json +++ b/package.json @@ -23,7 +23,7 @@ "build:deno": "node transpile.deno.js", "build:cf": "node transpile.cf.js", "test": "npm run test:esm && npm run test:cjs && npm run test:deno", - "test:esm": "node tests/index.js", + "test:esm": "node tests/index.js && node tests/reject_throttle.js", "test:cjs": "npm run build:cjs && cd cjs/tests && node index.js && cd ../../", "test:deno": "npm run build:deno && cd deno/tests && deno run --no-lock --allow-all --unsafely-ignore-certificate-errors index.js && cd ../../", "lint": "eslint src && eslint tests", diff --git a/src/connection.js b/src/connection.js index 1b1cccde..f27701c0 100644 --- a/src/connection.js +++ b/src/connection.js @@ -49,7 +49,7 @@ const errorFields = { 82 : 'routine' // R } -function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose = noop } = {}) { +function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose = noop, requeue = noop } = {}) { const { sslnegotiation, ssl, @@ -362,6 +362,58 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose setTimeout(connect, closedTime ? Math.max(0, closedTime + delay - performance.now()) : 0) } + function backOffOnError() { + // Escalate the shared backoff on a real error — the increment closed() does on a + // drop. Callers recompute `delay` from it unconditionally (as the original does), + // so only this escalation is gated: called `hadError && ...` on the drop path, and + // unconditionally from our establishment reconnect (a close there is always an error). + options.shared.retries++ + } + + function reconnect_with_reject_throttle() { + const t = options.shared.throttle + if (t.prober === null || t.prober === connection) + prober_reconnect() + else + stand_down() + } + + function prober_reconnect() { + // Become / stay the lone prober and pace the retry exactly as closed() paces a + // drop; every other establishing connection blocks behind us until we hand off. + // closedTime is set by closed() (as for a drop) — reconnect() only reads it. + options.shared.throttle.prober = connection + backOffOnError() + delay = (typeof backoff === 'function' ? backoff(options.shared.retries) : backoff) * 1000 + reconnect() + } + + function stand_down() { + // A non-prober open failed (a slow-start admit): collapse the ramp and give up + // this slot — hand the query back to the pool instead of retrying (which would + // re-form the herd). The prober keeps probing, now paced by the higher retries. + const t = options.shared.throttle + t.ramp = 1 + t.allowance = 0 + requeue(initial) + initial = null + onclose(connection, Errors.connection('CONNECTION_CLOSED', options, socket)) + } + + function throttle_on_success() { + // Only the prober's success clocks a recovery round: refill the admit allowance + // with the current batch and double the ramp (capped at the pool max). We do NOT + // null the prober here — it stays engaged (as this now-healthy connection) until + // throttleAdmit hands the baton to the next waiter, or to null once the backlog + // is drained. Admit successes are silent, else N concurrent successes compound it. + const t = options.shared.throttle + if (!options.reject_throttle || t.prober !== connection) + return + + t.allowance += t.ramp + t.ramp = Math.min(t.ramp * 2, options.max) + } + function connected() { try { statements = {} @@ -447,12 +499,16 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose socket.removeAllListeners() socket = null - if (initial) - return reconnect() + if (initial) { + if (!options.reject_throttle) + return reconnect() + closedTime = performance.now() + return reconnect_with_reject_throttle() + } !hadError && (query || sent.length) && error(Errors.connection('CONNECTION_CLOSED', options, socket)) closedTime = performance.now() - hadError && options.shared.retries++ + hadError && backOffOnError() delay = (typeof backoff === 'function' ? backoff(options.shared.retries) : backoff) * 1000 onclose(connection, Errors.connection('CONNECTION_CLOSED', options, socket)) } @@ -566,6 +622,7 @@ function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose initial && !initial.reserve && execute(initial) options.shared.retries = retries = 0 + throttle_on_success() initial = null return } diff --git a/src/index.js b/src/index.js index c7fba3da..b48910b2 100644 --- a/src/index.js +++ b/src/index.js @@ -62,7 +62,7 @@ function Postgres(a, b) { , full = Queue() , queues = { connecting, reserved, closed, ended, open, busy, full } - const connections = [...Array(options.max)].map(() => Connection(options, queues, { onopen, onend, onclose })) + const connections = [...Array(options.max)].map(() => Connection(options, queues, { onopen, onend, onclose, requeue })) const sql = Sql(handler) @@ -207,7 +207,7 @@ function Postgres(a, b) { : await new Promise((resolve, reject) => { const query = { reserve: resolve, reject } queries.push(query) - closed.length && connect(closed.shift(), query) + closed.length && throttleAllows() && connect(closed.shift(), query) }) move(c, reserved) @@ -333,7 +333,7 @@ function Postgres(a, b) { if (open.length) return go(open.shift(), query) - if (closed.length) + if (closed.length && throttleAllows()) return connect(closed.shift(), query) busy.length @@ -388,19 +388,69 @@ function Postgres(a, b) { resolve() } + // reject_throttle (opt-in): after a connection fails to ESTABLISH, block new + // opens (closed -> connecting) so the pool doesn't stampede a server that is + // rejecting connections. Complements `backoff` — backoff only *delays* each + // retry, and since its counter is pool-shared every connection waits the same + // delay and then fires together (a delayed herd); this instead keeps one failed + // connection as the lone prober while the rest block, then admits opens on an + // exponentially-growing `allowance` that each prober handshake-success refills + // (slow-start), spending one per admitted open. Returns true when a new open may + // proceed now (always, when the feature is off or the breaker is idle). + function throttleAllows() { + const t = options.shared.throttle + if (!options.reject_throttle || t.prober === null) + return true + if (t.allowance > 0) + return t.allowance--, true + return false + } + function connect(c, query) { move(c, connecting) c.connect(query) return c } + // reject_throttle: hand a query pinned to a stood-down connection back to the pool + // (an idle connection serves it, otherwise it waits in the backlog) rather than + // having that connection retry — which is what would re-form the herd. + function requeue(query) { + query.reserve ? queries.push(query) : handler(query) + } + + // reject_throttle: a prober's handshake success refilled `allowance` with the next + // (doubled) slow-start batch. Open up to that many for the backlog and make the last + // one opened the new prober, so exactly one connection keeps probing while the + // breaker stays engaged. Backlog dry => nothing opened, prober stays null => the + // breaker disengages. + function throttleAdmit() { + const t = options.shared.throttle + if (!options.reject_throttle || t.allowance === 0) + return + let last = null + while (t.allowance > 0 && queries.length && closed.length) { + t.allowance-- + last = connect(closed.shift(), queries.shift()) + } + if (last) { + t.prober = last // hand the baton to the last waiter opened; breaker stays engaged + } else { + t.prober = null // nothing left to admit => recovered => disengage, reset to baseline + t.ramp = 1 + t.allowance = 0 + } + } + function onend(c) { move(c, ended) } function onopen(c) { - if (queries.length === 0) + if (queries.length === 0) { + throttleAdmit() // no backlog: disengage the breaker if a prober success granted allowance return move(c, open) + } let max = Math.ceil(queries.length / (connecting.length + 1)) , ready = true @@ -416,6 +466,8 @@ function Postgres(a, b) { ready ? move(c, busy) : move(c, full) + + throttleAdmit() } function onclose(c, e) { @@ -423,7 +475,7 @@ function Postgres(a, b) { c.reserved = null c.onclose && (c.onclose(e), c.onclose = null) options.onclose && options.onclose(c.id) - queries.length && connect(c, queries.shift()) + queries.length && throttleAllows() && connect(c, queries.shift()) } } @@ -455,6 +507,7 @@ function parseOptions(a, b) { max_pipeline : 100, backoff : backoff, keep_alive : 60, + reject_throttle : false, prepare : true, debug : false, fetch_types : true, @@ -495,7 +548,7 @@ function parseOptions(a, b) { socket : o.socket, transform : parseTransform(o.transform || { undefined: undefined }), parameters : {}, - shared : { retries: 0, typeArrayMap: {} }, + shared : { retries: 0, typeArrayMap: {}, throttle: { prober: null, ramp: 1, allowance: 0 } }, ...mergeUserTypes(o.types) } } diff --git a/tests/reject_throttle.js b/tests/reject_throttle.js new file mode 100644 index 00000000..f90ef964 --- /dev/null +++ b/tests/reject_throttle.js @@ -0,0 +1,329 @@ +import { t } from './test.js' // eslint-disable-line +import net from 'net' + +import postgres from '../src/index.js' + +const delay = ms => new Promise(r => setTimeout(r, ms)) + +// The real test database (bootstrapped by tests/bootstrap.js). +const REAL = { host: '127.0.0.1', port: 5432, db: 'postgres_js_test', user: 'postgres_js_test' } + +// A controllable stand-in for the database endpoint. +// +// It records the timestamp of every inbound connection attempt, and for the first +// `rejectFirst` of them it reads the client's startup packet and then closes the +// socket CLEANLY (FIN) — which is what drives postgres.js down the establishment +// *reconnect* path (`if (initial) return reconnect()`), the path reject_throttle +// governs. (A destroy()/RST would instead take the error path and reject the query +// outright, never reaching the reconnect logic.) Once past `rejectFirst`, it +// transparently proxies to the real postgres so genuine handshakes complete and the +// breaker can recover. +function endpoint({ rejectFirst = Infinity } = {}) { + const attempts = [] + const sockets = new Set() + const server = net.createServer(client => { + attempts.push(Date.now()) + sockets.add(client) + client.on('error', () => {}) + client.on('close', () => sockets.delete(client)) + if (attempts.length <= rejectFirst) { + client.on('data', () => client.end()) // clean FIN after startup => establishment reconnect + return + } + // Proxy to the real database. Track the upstream socket alongside the client one: + // it is a live pg backend, and if close() only tears down the client side the + // upstream can stay ESTABLISHED, keeping the event loop alive long after the + // suite finishes (the harness relies on a natural exit) — an occasional CI hang. + const up = net.connect(REAL.port, REAL.host) + sockets.add(up) + up.on('error', () => client.destroy()) + up.on('close', () => sockets.delete(up)) + up.pipe(client) + client.pipe(up) + }) + return { + attempts, + listen: () => new Promise(r => server.listen(0, '127.0.0.1', () => r(server.address().port))), + close: () => { + sockets.forEach(s => s.destroy()) + return new Promise(r => server.close(r)) + } + } +} + +const opts = (port, extra) => ({ + host: '127.0.0.1', + port, + db: REAL.db, + user: REAL.user, + connect_timeout: 1, + ...extra +}) + +// --------------------------------------------------------------------------- +// Option + state plumbing +// --------------------------------------------------------------------------- + +t('reject_throttle defaults to false', async() => + [false, postgres({ max: 1 }).options.reject_throttle] +) + +t('reject_throttle can be enabled via options', async() => + [true, postgres({ max: 1, reject_throttle: true }).options.reject_throttle] +) + +t('reject_throttle reads from the connection string', async() => { + // Non-int query params stay strings in postgres.js (as for prepare/ssl/…): 'true' + // is truthy so the feature enables, and 'false' is coerced to boolean false. + const on = postgres('postgres://localhost/db?reject_throttle=true').options.reject_throttle + const off = postgres('postgres://localhost/db?reject_throttle=false').options.reject_throttle + return ['true,false', [!!on, off].join(',')] +}) + +t('throttle state is initialised on shared', async() => { + const th = postgres({ max: 1 }).options.shared.throttle + return ['true,1,0', [th.prober === null, th.ramp, th.allowance].join(',')] +}) + +// --------------------------------------------------------------------------- +// The core: an establishment-failure storm is throttled when enabled, and is a +// herd when not. +// --------------------------------------------------------------------------- + +t('off: establishment failures herd (hundreds of attempts)', { timeout: 10 }, async() => { + const ep = endpoint() // always rejects (clean FIN) + const port = await ep.listen() + const sql = postgres(opts(port, { max: 5, reject_throttle: false })) + for (let i = 0; i < 5; i++) sql`select 1`.catch(() => {}) + await delay(400) + const attempts = ep.attempts.length + const engaged = sql.options.shared.throttle.prober !== null + await sql.end({ timeout: 0 }).catch(() => {}) + await ep.close() + // vanilla retries each of the 5 connections immediately => a flood; breaker idle + return ['true,false', [attempts > 100, engaged].join(',')] +}) + +t('on: establishment failures are throttled to a trickle', { timeout: 10 }, async() => { + const ep = endpoint() // always rejects + const port = await ep.listen() + const sql = postgres(opts(port, { max: 5, reject_throttle: true })) + for (let i = 0; i < 5; i++) sql`select 1`.catch(() => {}) + await delay(400) + const attempts = ep.attempts.length + const engaged = sql.options.shared.throttle.prober !== null + await sql.end({ timeout: 0 }).catch(() => {}) + await ep.close() + // initial burst (<=max) + a few paced prober retries; two orders of magnitude fewer + return ['true,true', [attempts < 30, engaged].join(',')] +}) + +t('on: a single prober keeps probing after the burst', { timeout: 10 }, async() => { + const ep = endpoint() + const port = await ep.listen() + const sql = postgres(opts(port, { max: 5, reject_throttle: true })) + for (let i = 0; i < 5; i++) sql`select 1`.catch(() => {}) + await delay(250) // let the burst + stand-downs settle + const before = ep.attempts.length + await delay(700) // one prober-retry window + const after = ep.attempts.length + await sql.end({ timeout: 0 }).catch(() => {}) + await ep.close() + // only the lone prober retries in this window — a small handful, not another burst + return [true, (after - before) <= 6] +}) + +t('on: ramp collapses to baseline while blocked', { timeout: 10 }, async() => { + const ep = endpoint() + const port = await ep.listen() + const sql = postgres(opts(port, { max: 5, reject_throttle: true })) + for (let i = 0; i < 5; i++) sql`select 1`.catch(() => {}) + await delay(300) + const th = sql.options.shared.throttle + const state = [th.ramp, th.allowance].join(',') + await sql.end({ timeout: 0 }).catch(() => {}) + await ep.close() + return ['1,0', state] +}) + +// --------------------------------------------------------------------------- +// Recovery: once the endpoint accepts again, queued work drains and the breaker +// disengages back to baseline. +// --------------------------------------------------------------------------- + +t('recovery: queries resolve once the endpoint accepts', { timeout: 15 }, async() => { + const ep = endpoint({ rejectFirst: 3 }) // reject 3, then proxy to real pg + const port = await ep.listen() + const sql = postgres(opts(port, { max: 3, reject_throttle: true })) + const results = await Promise.all([sql`select 1 as x`, sql`select 2 as x`, sql`select 3 as x`]) + .then(rs => rs.map(r => r[0].x).join(',')) + .catch(e => 'ERR:' + (e.code || e.message)) + await sql.end({ timeout: 0 }).catch(() => {}) + await ep.close() + return ['1,2,3', results] +}) + +t('recovery: breaker disengages to baseline after draining', { timeout: 15 }, async() => { + const ep = endpoint({ rejectFirst: 4 }) + const port = await ep.listen() + const sql = postgres(opts(port, { max: 3, reject_throttle: true })) + await Promise.all([sql`select 1`, sql`select 2`, sql`select 3`]).catch(() => {}) + await delay(150) // let the final onopen run + const th = sql.options.shared.throttle + const state = [th.prober === null, th.ramp, th.allowance].join(',') + await sql.end({ timeout: 0 }).catch(() => {}) + await ep.close() + return ['true,1,0', state] +}) + +t('recovery: works after a large reject streak', { timeout: 20 }, async() => { + const ep = endpoint({ rejectFirst: 10 }) + const port = await ep.listen() + const sql = postgres(opts(port, { max: 5, reject_throttle: true })) + const ok = await Promise.all(Array.from({ length: 5 }, (_, i) => sql`select ${i} as x`)) + .then(rs => rs.map(r => r[0].x).join(',')) + .catch(e => 'ERR:' + (e.code || e.message)) + await delay(150) + const disengaged = sql.options.shared.throttle.prober === null + await sql.end({ timeout: 0 }).catch(() => {}) + await ep.close() + return ['0,1,2,3,4|true', ok + '|' + disengaged] +}) + +// --------------------------------------------------------------------------- +// Feature-off is byte-for-byte vanilla against a healthy endpoint. +// --------------------------------------------------------------------------- + +t('off: normal queries against a healthy endpoint are unaffected', async() => { + const sql = postgres({ ...REAL, max: 1, reject_throttle: false }) + const x = (await sql`select 42 as x`)[0].x + await sql.end({ timeout: 0 }).catch(() => {}) + return [42, x] +}) + +t('on: normal queries against a healthy endpoint still work', async() => { + const sql = postgres({ ...REAL, max: 1, reject_throttle: true }) + const x = (await sql`select 42 as x`)[0].x + const idle = sql.options.shared.throttle.prober === null + await sql.end({ timeout: 0 }).catch(() => {}) + return ['42,true', [x, idle].join(',')] +}) + +// --------------------------------------------------------------------------- +// More coverage +// --------------------------------------------------------------------------- + +t('off: never touches the throttle state', { timeout: 10 }, async() => { + const ep = endpoint() + const port = await ep.listen() + const sql = postgres(opts(port, { max: 5, reject_throttle: false })) + for (let i = 0; i < 5; i++) sql`select 1`.catch(() => {}) + await delay(300) + const th = sql.options.shared.throttle + const state = [th.prober === null, th.ramp, th.allowance].join(',') + await sql.end({ timeout: 0 }).catch(() => {}) + await ep.close() + return ['true,1,0', state] +}) + +t('on: a backlog far exceeding max is still throttled', { timeout: 10 }, async() => { + const ep = endpoint() + const port = await ep.listen() + const sql = postgres(opts(port, { max: 3, reject_throttle: true })) + for (let i = 0; i < 30; i++) sql`select 1`.catch(() => {}) + await delay(400) + const attempts = ep.attempts.length + await sql.end({ timeout: 0 }).catch(() => {}) + await ep.close() + // 30 queued queries, but only <=max open at once + the lone prober's retries + return [true, attempts < 30] +}) + +t('on: max:1 recovers through the lone prober', { timeout: 15 }, async() => { + const ep = endpoint({ rejectFirst: 3 }) + const port = await ep.listen() + const sql = postgres(opts(port, { max: 1, reject_throttle: true })) + const x = await sql`select 7 as x`.then(r => r[0].x).catch(e => 'ERR:' + (e.code || e.message)) + await delay(150) + const disengaged = sql.options.shared.throttle.prober === null + await sql.end({ timeout: 0 }).catch(() => {}) + await ep.close() + return ['7,true', [x, disengaged].join(',')] +}) + +t('recovery: drains a large backlog and disengages', { timeout: 20 }, async() => { + const ep = endpoint({ rejectFirst: 5 }) + const port = await ep.listen() + const sql = postgres(opts(port, { max: 4, reject_throttle: true })) + const results = await Promise.all(Array.from({ length: 12 }, (_, i) => sql`select ${i} as x`)) + .then(rs => rs.map(r => r[0].x).join(',')) + .catch(e => 'ERR:' + (e.code || e.message)) + await delay(200) + const disengaged = sql.options.shared.throttle.prober === null + await sql.end({ timeout: 0 }).catch(() => {}) + await ep.close() + return ['0,1,2,3,4,5,6,7,8,9,10,11|true', results + '|' + disengaged] +}) + +t('on: max_lifetime recycle still reconnects (drop path intact)', { timeout: 10 }, async() => { + const sql = postgres({ ...REAL, max: 1, reject_throttle: true, max_lifetime: 0.1, idle_timeout: 1 }) + const a = (await sql`select 1 as x`)[0].x + await delay(300) // exceed max_lifetime => connection recycles (drop path) + const b = (await sql`select 2 as x`)[0].x + const idle = sql.options.shared.throttle.prober === null + await sql.end({ timeout: 0 }).catch(() => {}) + return ['1,2,true', [a, b, idle].join(',')] +}) + +// --------------------------------------------------------------------------- +// Boundaries: the breaker governs ONLY the clean-close establishment-reconnect +// path. Error-close (RST) and connect-timeout must still reject, not hang. +// --------------------------------------------------------------------------- + +t('on: an establishment connection error still rejects (not the reconnect path)', { timeout: 10 }, async() => { + // Bind then immediately release a port so connecting to it yields ECONNREFUSED — + // a genuine establishment *error* (socket 'error' => errored() rejects and clears + // `initial`, so closed() never takes the reconnect branch), which the breaker must + // pass straight through. A server-side destroy() is deliberately NOT used here: it + // isn't portable, because on Linux the socket is often torn down before the client's + // startup packet lands, which the client sees as a clean mid-handshake close => the + // throttled reconnect path => the query never settles. ECONNREFUSED is deterministic + // across OSes and Node versions. + const probe = net.createServer() + const port = await new Promise(r => probe.listen(0, '127.0.0.1', () => r(probe.address().port))) + await new Promise(r => probe.close(r)) + const sql = postgres(opts(port, { max: 1, reject_throttle: true })) + const settled = await sql`select 1`.then(() => 'RESOLVED', e => e.code || 'ERR') + await sql.end({ timeout: 0 }).catch(() => {}) + // it must settle (with an error) rather than hang in a reconnect loop + return [true, settled !== 'RESOLVED'] +}) + +t('on: connect_timeout still fires (not swallowed by the breaker)', { timeout: 10 }, async() => { + const socks = new Set() + const server = net.createServer(c => (socks.add(c), c.on('close', () => socks.delete(c)))) // accepts, never responds + const port = await new Promise(r => server.listen(0, '127.0.0.1', () => r(server.address().port))) + const sql = postgres(opts(port, { max: 1, reject_throttle: true, connect_timeout: 0.3 })) + const code = await sql`select 1`.then(() => 'RESOLVED', e => e.code) + await sql.end({ timeout: 0 }).catch(() => {}) + socks.forEach(s => s.destroy()) + await new Promise(r => server.close(r)) + return ['CONNECT_TIMEOUT', code] +}) + +t('on drives ~orders-of-magnitude fewer attempts than off', { timeout: 12 }, async() => { + async function count(reject_throttle) { + const ep = endpoint() + const port = await ep.listen() + const sql = postgres(opts(port, { max: 5, reject_throttle })) + for (let i = 0; i < 5; i++) sql`select 1`.catch(() => {}) + await delay(400) + const n = ep.attempts.length + await sql.end({ timeout: 0 }).catch(() => {}) + await ep.close() + return n + } + const off = await count(false) + const on = await count(true) + return [true, off > on * 10] +})