Add opt-in reject_throttle: pace connection opens after establishment failures

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",

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
     }

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,14 +466,16 @@ function Postgres(a, b) {
     ready
       ? move(c, busy)
       : move(c, full)
+
+    throttleAdmit()
   }
 
   function onclose(c, e) {
     move(c, closed)
     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)
   }
 }