Improve connection queue handling + fix leak

Author: porsager

src/connection.js

@@ -48,7 +48,7 @@ const errorFields = {
   82  : 'routine'            // R
 }
 
-function Connection(options, { onopen = noop, onend = noop, ondrain = noop, onclose = noop } = {}) {
+function Connection(options, queues = {}, { onopen = noop, onend = noop, onclose = noop } = {}) {
   const {
     ssl,
     max,
@@ -80,7 +80,6 @@ function Connection(options, { onopen = noop, onend = noop, ondrain = noop, oncl
     , needsTypes = options.fetch_types
     , backendParameters = {}
     , statements = {}
-    , state = 'closed'
     , statementId = Math.random().toString(36).slice(2)
     , statementCount = 1
     , closedDate = 0
@@ -105,13 +104,8 @@ function Connection(options, { onopen = noop, onend = noop, ondrain = noop, oncl
     , final = null
 
   const connection = {
-    get state() { return state },
-    set state(x) {
-      state = x
-      state === 'open'
-        ? idleTimer.start()
-        : idleTimer.cancel()
-    },
+    queue: queues.closed,
+    idleTimer,
     connect(query) {
       initial = query
       reconnect()
@@ -124,6 +118,8 @@ function Connection(options, { onopen = noop, onend = noop, ondrain = noop, oncl
     id
   }
 
+  queues.closed && queues.closed.push(connection)
+
   return connection
 
   function createSocket() {
@@ -291,7 +287,7 @@ function Connection(options, { onopen = noop, onend = noop, ondrain = noop, oncl
 
   /* c8 ignore next 3 */
   function drain() {
-    ondrain(connection)
+    onopen(connection)
   }
 
   function data(x) {
@@ -362,7 +358,7 @@ function Connection(options, { onopen = noop, onend = noop, ondrain = noop, oncl
   }
 
   function error(err) {
-    if (connection.state === 'connecting' && options.host[retries + 1])
+    if (connection.queue === queues.connecting && options.host[retries + 1])
       return
 
     errored(err)
@@ -529,7 +525,7 @@ function Connection(options, { onopen = noop, onend = noop, ondrain = noop, oncl
     }
 
     while (sent.length && (query = sent.shift()) && (query.active = true) && query.cancelled)
-      Connection(options, {}).cancel(query.state, query.cancelled.resolve, query.cancelled.reject)
+      Connection(options).cancel(query.state, query.cancelled.resolve, query.cancelled.reject)
 
     if (query)
       return // Consider opening if able and sent.length < 50

src/index.js

@@ -42,15 +42,16 @@ function Postgres(a, b) {
   let ending = false
 
   const queries = Queue()
-      , connections = [...Array(options.max)].map(() => Connection(options, { onopen, onend, ondrain, onclose }))
-      , closed = Queue(connections)
+      , connecting = Queue()
       , reserved = Queue()
+      , closed = Queue()
+      , ended = Queue()
       , open = Queue()
       , busy = Queue()
       , full = Queue()
-      , ended = Queue()
-      , connecting = Queue()
-      , queues = { closed, ended, connecting, reserved, open, busy, full }
+      , queues = { connecting, reserved, closed, ended, open, busy, full }
+
+  const connections = [...Array(options.max)].map(() => Connection(options, queues, { onopen, onend, onclose }))
 
   const sql = Sql(handler)
 
@@ -229,23 +230,30 @@ function Postgres(a, b) {
 
       function handler(q) {
         q.catch(e => uncaughtError || (uncaughtError = e))
-        c.state === 'full'
+        c.queue === full
           ? queries.push(q)
-          : c.execute(q) || (c.state = 'full', full.push(c))
+          : c.execute(q) || move(c, full)
       }
     }
 
     function onexecute(c) {
-      queues[c.state].remove(c)
-      c.state = 'reserved'
+      connection = c
+      move(c, reserved)
       c.reserved = () => queries.length
         ? c.execute(queries.shift())
-        : c.state = 'reserved'
-      reserved.push(c)
-      connection = c
+        : move(c, reserved)
     }
   }
 
+  function move(c, queue) {
+    c.queue.remove(c)
+    queue.push(c)
+    c.queue = queue
+    queue === open
+      ? c.idleTimer.start()
+      : c.idleTimer.cancel()
+  }
+
   function json(x) {
     return new Parameter(x, 3802)
   }
@@ -262,28 +270,27 @@ function Postgres(a, b) {
       return query.reject(Errors.connection('CONNECTION_ENDED', options, options))
 
     if (open.length)
-      return go(open, query)
+      return go(open.shift(), query)
 
     if (closed.length)
       return connect(closed.shift(), query)
 
     busy.length
-      ? go(busy, query)
+      ? go(busy.shift(), query)
       : queries.push(query)
   }
 
-  function go(xs, query) {
-    const c = xs.shift()
+  function go(c, query) {
     return c.execute(query)
-      ? (c.state = 'busy', busy.push(c))
-      : (c.state = 'full', full.push(c))
+      ? move(c, busy)
+      : move(c, full)
   }
 
   function cancel(query) {
     return new Promise((resolve, reject) => {
       query.state
         ? query.active
-          ? Connection(options, {}).cancel(query.state, resolve, reject)
+          ? Connection(options).cancel(query.state, resolve, reject)
           : query.cancelled = { resolve, reject }
         : (
           queries.remove(query),
@@ -317,21 +324,17 @@ function Postgres(a, b) {
   }
 
   function connect(c, query) {
-    c.state = 'connecting'
-    connecting.push(c)
+    move(c, connecting)
     c.connect(query)
   }
 
   function onend(c) {
-    queues[c.state].remove(c)
-    c.state = 'ended'
-    ended.push(c)
+    move(c, ended)
   }
 
   function onopen(c) {
-    queues[c.state].remove(c)
     if (queries.length === 0)
-      return (c.state = 'open', open.push(c))
+      return move(c, open)
 
     let max = Math.ceil(queries.length / (connecting.length + 1))
       , ready = true
@@ -340,23 +343,15 @@ function Postgres(a, b) {
       ready = c.execute(queries.shift())
 
     ready
-      ? (c.state = 'busy', busy.push(c))
-      : (c.state = 'full', full.push(c))
-  }
-
-  function ondrain(c) {
-    full.remove(c)
-    onopen(c)
+      ? move(c, busy)
+      : move(c, full)
   }
 
   function onclose(c) {
-    queues[c.state].remove(c)
-    c.state = 'closed'
+    move(c, closed)
     c.reserved = null
     options.onclose && options.onclose(c.id)
-    queries.length
-      ? connect(c, queries.shift())
-      : queues.closed.push(c)
+    queries.length && connect(c, queries.shift())
   }
 }