Close session and connection used for rediscovery

Previously session and the underlying connection were not returned to the pool
after the rediscovery procedure. This happened because rediscovery session was
not initalized with an appropriate callback.

This commit fixes the problem by extracting the release callback code into a
function that gets called for both "normal" and "rediscovery" sessions.
Probably a cleaner way to fix this would be to move functionality into
`Session#close()` and make connection release itself to the pool. However
this resulted into lots of non-trivial changes and thus should go into a
separate PR.

Author: lutovich

src/v1/driver.js

@@ -119,30 +119,33 @@ class Driver {
         //we don't need to tell the driver about this error
       }
     });
-    return this._createSession(connectionPromise, (cb) => {
-      // This gets called on Session#close(), and is where we return
-      // the pooled 'connection' instance.
-
-      // We don't pool Session instances, to avoid users using the Session
-      // after they've called close. The `Session` object is just a thin
-      // wrapper around Connection anyway, so it makes little difference.
-
-      // Queue up a 'reset', to ensure the next user gets a clean
-      // session to work with.
+    return this._createSession(connectionPromise, this._releaseConnection(connectionPromise));
+  }
 
-      connectionPromise.then( (conn) => {
+  /**
+   * The returned function gets called on Session#close(), and is where we return the pooled 'connection' instance.
+   * We don't pool Session instances, to avoid users using the Session after they've called close.
+   * The `Session` object is just a thin wrapper around Connection anyway, so it makes little difference.
+   * @param {Promise} connectionPromise - promise resolved with the connection.
+   * @return {function(*=)} - function that releases the connection and then executes an optional callback.
+   * @protected
+   */
+  _releaseConnection(connectionPromise) {
+    return userDefinedCallback => {
+      connectionPromise.then(conn => {
+        // Queue up a 'reset', to ensure the next user gets a clean session to work with.
         conn.reset();
         conn.sync();
 
         // Return connection to the pool
         conn._release();
-      }).catch( () => {/*ignore errors here*/});
+      }).catch(ignoredError => {
+      });
 
-      // Call user callback
-      if (cb) {
-        cb();
+      if (userDefinedCallback) {
+        userDefinedCallback();
       }
-    });
+    };
   }
 
   static _validateSessionMode(rawMode) {

src/v1/internal/get-servers-util.js

@@ -27,8 +27,8 @@ const PROCEDURE_NOT_FOUND_CODE = 'Neo.ClientError.Procedure.ProcedureNotFound';
 export default class GetServersUtil {
 
   callGetServers(session, routerAddress) {
-    // todo: session should be closed here along with the underlying connection
     return session.run(PROCEDURE_CALL).then(result => {
+      session.close();
       return result.records;
     }).catch(error => {
       if (error.code === PROCEDURE_NOT_FOUND_CODE) {

src/v1/routing-driver.js

@@ -129,9 +129,9 @@ class RoutingDriver extends Driver {
           this._forget(previousRouter);
         }
 
-        // todo: properly close this connection
         const connection = this._pool.acquire(currentRouter);
-        const session = this._createSession(Promise.resolve(connection));
+        const connectionPromise = Promise.resolve(connection);
+        const session = this._createSession(connectionPromise, this._releaseConnection(connectionPromise));
 
         // try next router
         return this._rediscovery.lookupRoutingTableOnRouter(session, currentRouter);

test/internal/get-servers-util.test.js

@@ -40,12 +40,30 @@ describe('get-servers-util', () => {
   it('should return retrieved records when query succeeds', done => {
     const session = FakeSession.successful({records: ['foo', 'bar', 'baz']});
 
-    callGetServers(session, '').then(records => {
+    callGetServers(session).then(records => {
       expect(records).toEqual(['foo', 'bar', 'baz']);
       done();
     }).catch(console.log);
   });
 
+  it('should close session when query succeeds', done => {
+    const session = FakeSession.successful({records: ['foo', 'bar', 'baz']});
+
+    callGetServers(session).then(() => {
+      expect(session.isClosed()).toBeTruthy();
+      done();
+    }).catch(console.log);
+  });
+
+  it('should not close session when query fails', done => {
+    const session = FakeSession.failed(newError('Oh no!', SESSION_EXPIRED));
+
+    callGetServers(session).then(() => {
+      expect(session.isClosed()).toBeFalsy();
+      done();
+    }).catch(console.log);
+  });
+
   it('should return null on connection error', done => {
     const session = FakeSession.failed(newError('Oh no!', SESSION_EXPIRED));
 
@@ -229,6 +247,7 @@ describe('get-servers-util', () => {
 
     constructor(runResponse) {
       this._runResponse = runResponse;
+      this._closed = false;
     }
 
     static successful(result) {
@@ -242,5 +261,13 @@ describe('get-servers-util', () => {
     run() {
       return this._runResponse;
     }
+
+    close() {
+      this._closed = true;
+    }
+
+    isClosed() {
+      return this._closed;
+    }
   }
 });

test/v1/routing.driver.boltkit.it.js

@@ -783,6 +783,66 @@ describe('routing driver', function () {
     });
   });
 
+  it('should close connection used for routing table refreshing', done => {
+    if (!boltkit.BoltKitSupport) {
+      done();
+      return;
+    }
+
+    const kit = new boltkit.BoltKit();
+    // server is both router and writer
+    const server = kit.start('./test/resources/boltkit/discover_new_servers.script', 9001);
+
+    kit.run(() => {
+      const driver = newDriver('bolt+routing://127.0.0.1:9001');
+
+      const acquiredConnections = [];
+      const releasedConnections = [];
+      setUpPoolToMemorizeAllAcquiredAndReleasedConnections(driver, acquiredConnections, releasedConnections);
+
+      const session = driver.session();
+      session.run('MATCH (n) RETURN n.name').then(() => {
+        session.close(() => {
+          driver.close();
+          server.exit(code => {
+            expect(code).toEqual(0);
+
+            // two connections should have been acquired: one for rediscovery and one for the query
+            expect(acquiredConnections.length).toEqual(2);
+            // same two connections should have been released
+            expect(releasedConnections.length).toEqual(2);
+
+            // verify that acquired connections are those that we released
+            for (let i = 0; i < acquiredConnections.length; i++) {
+              expect(acquiredConnections[i]).toBe(releasedConnections[i]);
+            }
+
+            done();
+          });
+        });
+      });
+    });
+  });
+
+  function setUpPoolToMemorizeAllAcquiredAndReleasedConnections(driver, acquiredConnections, releasedConnections) {
+    // make connection pool remember all acquired connections
+    const originalAcquire = driver._pool.acquire.bind(driver._pool);
+    const memorizingAcquire = (...args) => {
+      const connection = originalAcquire(...args);
+      acquiredConnections.push(connection);
+      return connection;
+    };
+    driver._pool.acquire = memorizingAcquire;
+
+    // make connection pool remember all released connections
+    const originalRelease = driver._pool._release;
+    const rememberingRelease = (key, resource) => {
+      originalRelease(key, resource);
+      releasedConnections.push(resource);
+    };
+    driver._pool._release = rememberingRelease;
+  }
+
   function newDriver(url) {
     // BoltKit currently does not support encryption, create driver with encryption turned off
     return neo4j.driver(url, neo4j.auth.basic("neo4j", "neo4j"), {