more tests! more everything.

Author: chrisdickinson

.gitignore

@@ -0,0 +1 @@
+coverage/

db-session.js

@@ -4,7 +4,6 @@ const DOMAIN_TO_SESSION = new WeakMap()
 const Promise = require('bluebird')
 const domain = require('domain')
 
-const AtomicSessionConnectionPair = require('./lib/atomic-session-connpair.js')
 const TxSessionConnectionPair = require('./lib/tx-session-connpair.js')
 const SessionConnectionPair = require('./lib/session-connpair.js')
 
@@ -43,17 +42,14 @@ const api = module.exports = {
   },
 
   getConnection () {
-    return DOMAIN_TO_SESSION.get(process.domain).getConnection()
+    return api.session.getConnection()
   },
 
   get session () {
     var current = DOMAIN_TO_SESSION.get(process.domain)
-    if (!current || !process.domain) {
+    if (!current || current.inactive || !process.domain) {
       throw new NoSessionAvailable()
     }
-    while (current.inactive && current.parent) {
-      current = current.parent
-    }
     return current
   },
 
@@ -93,15 +89,20 @@ class Session {
 
   transaction (operation, args) {
     const getConnPair = this.getConnection()
-    const getResult = Session$RunWrapped(this, connPair => {
-      return new TransactionSession(this, connPair)
+    const getResult = Session$RunWrapped(connPair => {
+      return new TransactionSession(connPair)
     }, getConnPair, `BEGIN`, {
       success: `COMMIT`,
       failure: `ROLLBACK`
     }, operation, args)
-    const releasePair = getResult.return(getConnPair).then(
-      pair => pair.release()
-    )
+
+    const releasePair = getConnPair.then(pair => {
+      return getResult.reflect().then(result => {
+        return result.isFulfilled()
+          ? pair.release()
+          : pair.release(result.reason())
+      })
+    })
 
     return releasePair.return(getResult)
   }
@@ -119,97 +120,107 @@ class Session {
 }
 
 class TransactionSession {
-  constructor (parent, connPair) {
-    this.parent = parent
+  constructor (connPair) {
     this.connectionPair = connPair
     this.inactive = false
     this.operation = Promise.resolve(true)
   }
 
   getConnection () {
     if (this.inactive) {
-      return this.parent.getConnection()
+      return new Promise((_, reject) => {
+        reject(new NoSessionAvailable())
+      })
     }
-    // XXX(chrisdickinson): creating a TxConnPair implicitly
+    // NB(chrisdickinson): creating a TxConnPair implicitly
     // swaps out "this.operation", creating a linked list of
     // promises.
     return new TxSessionConnectionPair(this).onready
   }
 
   transaction (operation, args) {
     if (this.inactive) {
-      return this.parent.transaction(operation, args)
+      return new Promise((_, reject) => {
+        reject(new NoSessionAvailable())
+      })
     }
     return operation.apply(null, args)
   }
 
   atomic (operation, args) {
-    const atomicConnPair = new AtomicSessionConnectionPair(this)
+    const atomicConnPair = this.getConnection()
     const savepointName = getSavepointName(operation)
-    const getResult = Session$RunWrapped(this, connPair => {
-      return new AtomicSession(this, connPair, savepointName)
-    }, atomicConnPair.onready, `SAVEPOINT ${savepointName}`, {
+    const getResult = Session$RunWrapped(connPair => {
+      return new AtomicSession(connPair, savepointName)
+    }, atomicConnPair, `SAVEPOINT ${savepointName}`, {
       success: `RELEASE SAVEPOINT ${savepointName}`,
       failure: `ROLLBACK TO SAVEPOINT ${savepointName}`
     }, operation, args)
 
-    return getResult.then(() => {
-      setImmediate(() => {
-        atomicConnPair.close()
+    const releasePair = atomicConnPair.then(pair => {
+      return getResult.reflect().then(result => {
+        return result.isFulfilled()
+          ? pair.release()
+          : pair.release(result.reason())
       })
-    }).return(getResult)
+    })
+
+    return releasePair.return(getResult)
   }
 }
 
 class AtomicSession extends TransactionSession {
-  constructor (parent, connection, name) {
-    super(parent, connection)
+  constructor (connection, name) {
+    super(connection)
     this.name = name
   }
 }
 
-function Session$RunWrapped (parent,
-                             createSession,
-                             getConnPair, before, after, operation, args) {
-  const createSubdomain = getConnPair.then(connPair => {
+function Session$RunWrapped (createSession,
+                             getConnPair,
+                             before,
+                             after,
+                             operation,
+                             args) {
+  return getConnPair.then(pair => {
     const subdomain = domain.create()
-    const session = createSession(connPair)
+    const session = createSession(pair)
     DOMAIN_TO_SESSION.set(subdomain, session)
-    return subdomain
-  })
 
-  const runBefore = getConnPair.then(connPair => new Promise(
-    (resolve, reject) => connPair.connection.query(
-      before,
-      err => err ? reject(err) : resolve()
-    )
-  ))
+    const runBefore = new Promise((resolve, reject) => {
+      return pair.connection.query(
+        before,
+        err => err ? reject(err) : resolve()
+      )
+    })
 
-  const getResult = runBefore.return(
-    createSubdomain
-  ).then(domain => {
-    args.unshift(operation)
-    return Promise.resolve(domain.run.apply(domain, args))
-  })
+    return runBefore.then(() => {
+      const opArgs = args.slice()
+      opArgs.unshift(operation)
+      const getResult = Promise.resolve(
+        subdomain.run.apply(subdomain, opArgs)
+      )
 
-  const getReflectedResult = getResult.reflect()
-  const runCommitStep = Promise.join(
-    getReflectedResult,
-    getConnPair.get('connection')
-  ).spread((result, connection) => {
-    return new Promise((resolve, reject) => {
-      connection.query(
-        result.isFulfilled()
-          ? after.success
-          : after.failure,
-        err => err ? reject(err) : resolve()
+      const waitOperation = Promise.join(
+        getResult,
+        getResult.then(() => session.operation)
       )
+      .finally(markInactive(subdomain))
+      .return(getResult.reflect())
+
+      const runCommitStep = waitOperation.then(result => {
+        return new Promise((resolve, reject) => {
+          return pair.connection.query(
+            result.isFulfilled()
+              ? after.success
+              : after.failure,
+            err => err ? reject(err) : resolve()
+          )
+        })
+      })
+      return runCommitStep.return(getResult)
     })
   })
-
-  return runCommitStep.return(
-    createSubdomain
-  ).then(markInactive(parent)).return(getResult)
 }
 
 function getSavepointName (operation) {
@@ -221,14 +232,11 @@ function getSavepointName (operation) {
 }
 getSavepointName.ID = 0
 
-function markInactive (session) {
-  return domain => {
-    domain.exit()
-    DOMAIN_TO_SESSION.get(domain).inactive = true
-
-    // if, somehow, we get a reference to this domain again, point
-    // it at the parent session.
-    DOMAIN_TO_SESSION.set(domain, session)
+function markInactive (subdomain) {
+  return () => {
+    subdomain.exit()
+    DOMAIN_TO_SESSION.get(subdomain).inactive = true
+    DOMAIN_TO_SESSION.set(subdomain, null)
   }
 }
 

lib/atomic-session-connpair.js

@@ -34,12 +34,11 @@ function release (conn, err) {
 // with the error notification; replay all pending connections
 // so they don't try to grab this one.
 function handleError (conn, err) {
-  conn.session.decrementConnections()
   conn.session.releasePair(conn.pair, err)
 
   const pending = conn.session.pending.slice()
-  conn.session.length = 0
-  while (pending.length) {
-    pending.shift().resolve(conn.session.getConnection())
-  }
+  conn.session.pending.length = 0
+  pending.forEach(xs => {
+    conn.session.getConnection().then(ys => xs.resolve(ys))
+  })
 }

lib/session-connpair.js

@@ -5,14 +5,39 @@ const Promise = require('bluebird')
 const RESOLVE_SYM = Symbol()
 const REJECT_SYM = Symbol()
 
+// while SessionConnectionPair (./session-connpair.js) represents
+// an *actual* postgres connection and "release" function,
+// TransactionSessionConnPair *wraps* a SessionConnectionPair in
+// order to serialize access to it. As such, "txPair.release()"
+// doesn't release the underlying connection, it just lets
+// subsequent operations run.
+//
+// the session machinery handles "fully" releasing the underlying
+// connection — see Session#transaction and TransactionSession#atomic
+// for more details (specifically, look for "release()".)
 module.exports = class TransactionSessionConnPair {
   constructor (session) {
     this.pair = session.connectionPair
     this.release = err => release(this, err)
     this.completed = new Promise((resolve, reject) => {
       this[RESOLVE_SYM] = resolve
       this[REJECT_SYM] = reject
+    }).catch((/* err */) => {
+      // XXX(chrisdickinson): this would be the place to
+      // add error monitoring.
     })
+
+    // the onready promise will let session methods know
+    // that previous operations have fully resolved —
+    // "session.operation" is "txPair.completed", so we end
+    // up with a linked list of promises, e.g.:
+    //
+    //                  duration of one transaction
+    //                  _ _ _ _ _ _ _|_ _ _ _ _ _ _
+    //                 /                           \
+    //                |                             |
+    //  completed → onready → (work happens) → completed → onready
+    //
     this.onready = session.operation.then(() => this)
     session.operation = this.completed
   }

lib/tx-session-connpair.js

@@ -4,8 +4,12 @@
   "description": "domain-attached database sessions",
   "main": "db-session.js",
   "scripts": {
-    "test": "tap test/*-test.js && standard",
-    "test-cov": "nyc tap test/*-test.js && standard"
+    "test:code": "tap test/*-test.js",
+    "test:style": "standard",
+    "test": "npm run test:code && npm run test:style",
+    "cov:test": "nyc npm run test:code",
+    "cov:html": "npm run cov:test && nyc report --reporter=lcov",
+    "cov:view": "npm run cov:html && ecstatic coverage/lcov-report -p 60888"
   },
   "repository": {
     "type": "git",
@@ -24,6 +28,7 @@
   },
   "homepage": "https://github.com/npm/pg-db-session#readme",
   "devDependencies": {
+    "ecstatic": "^1.4.0",
     "nyc": "^5.3.0",
     "pg": "^4.4.3",
     "standard": "^5.4.1",