Pool: separate validation on acquire and on release and add the acquisition context to create and validate

Author: bigmontz

packages/bolt-connection/src/connection-provider/connection-provider-direct.js

@@ -49,7 +49,7 @@ export default class DirectConnectionProvider extends PooledConnectionProvider {
         this._handleAuthorizationExpired(error, address, conn, database)
     })
 
-    const connection = await this._connectionPool.acquire(this._address)
+    const connection = await this._connectionPool.acquire({}, this._address)
 
     if (auth && auth !== connection.authToken) {
       if (connection.supportsReAuth) {

packages/bolt-connection/src/connection-provider/connection-provider-pooled.js

@@ -47,7 +47,8 @@ export default class PooledConnectionProvider extends ConnectionProvider {
     this._connectionPool = new Pool({
       create: this._createConnection.bind(this),
       destroy: this._destroyConnection.bind(this),
-      validate: this._validateConnection.bind(this),
+      validateOnAcquire: this._validateConnectionOnAcquire.bind(this),
+      validateOnRelease: this._validateConnectionOnRelease.bind(this),
       installIdleObserver: PooledConnectionProvider._installIdleObserverOnConnection.bind(
         this
       ),
@@ -69,7 +70,7 @@ export default class PooledConnectionProvider extends ConnectionProvider {
    * @return {Promise<Connection>} promise resolved with a new connection or rejected when failed to connect.
    * @access private
    */
-  _createConnection (address, release) {
+  _createConnection (_context, address, release) {
     return this._createChannelConnection(address).then(connection => {
       connection._release = () => {
         return release(address, connection)
@@ -85,12 +86,32 @@ export default class PooledConnectionProvider extends ConnectionProvider {
     })
   }
 
+  async _validateConnectionOnAcquire (_context, conn) {
+    if (!this._validateConnection(conn)) {
+      return false
+    }
+
+    try {
+      await this._authenticationProvider.authenticate({ connection: conn })
+      return true
+    } catch (error) {
+      this._log.info(
+        `The connection ${conn.id} is not valid because of an error ${error.code} '${error.message}'`
+      )
+      return false
+    }
+  }
+
+  _validateConnectionOnRelease (conn) {
+    return this._validateConnection(conn)
+  }
+
   /**
    * Check that a connection is usable
    * @return {boolean} true if the connection is open
    * @access private
    **/
-  async _validateConnection (conn) {
+  _validateConnection (conn) {
     if (!conn.isOpen()) {
       return false
     }
@@ -101,15 +122,6 @@ export default class PooledConnectionProvider extends ConnectionProvider {
       return false
     }
 
-    try {
-      await this._authenticationProvider.authenticate({ connection: conn })
-    } catch (error) {
-      this._log.info(
-        `The connection ${conn.id} is not valid because of an error ${error.code} '${error.message}'`
-      )
-      return false
-    }
-
     return true
   }
 
@@ -130,7 +142,7 @@ export default class PooledConnectionProvider extends ConnectionProvider {
    * @return {Promise<ServerInfo>} the server info
    */
   async _verifyConnectivityAndGetServerVersion ({ address }) {
-    const connection = await this._connectionPool.acquire(address)
+    const connection = await this._connectionPool.acquire({}, address)
     const serverInfo = new ServerInfo(connection.server, connection.protocol().version)
     try {
       if (!connection.protocol().isLastMessageLogin()) {

packages/bolt-connection/src/connection-provider/connection-provider-routing.js

@@ -317,7 +317,7 @@ export default class RoutingConnectionProvider extends PooledConnectionProvider
   }
 
   _acquireConnectionToServer (address, serverName, routingTable) {
-    return this._connectionPool.acquire(address)
+    return this._connectionPool.acquire({}, address)
   }
 
   _freshRoutingTable ({ accessMode, database, bookmarks, impersonatedUser, onDatabaseNameResolved, auth } = {}) {
@@ -559,7 +559,7 @@ export default class RoutingConnectionProvider extends PooledConnectionProvider
 
   async _createSessionForRediscovery (routerAddress, bookmarks, impersonatedUser, auth) {
     try {
-      let connection = await this._connectionPool.acquire(routerAddress)
+      let connection = await this._connectionPool.acquire({}, routerAddress)
 
       if (auth && connection.authToken !== auth) {
         if (connection.supportsReAuth) {
@@ -767,7 +767,6 @@ function _isFailFastError (error) {
 }
 
 function _isFailFastSecurityError (error) {
-  console.error(error)
   return error.code.startsWith('Neo.ClientError.Security.') &&
     ![
       AUTHORIZATION_EXPIRED_CODE

packages/bolt-connection/src/pool/pool.js

@@ -26,15 +26,18 @@ const {
 
 class Pool {
   /**
-   * @param {function(address: ServerAddress, function(address: ServerAddress, resource: object): Promise<object>): Promise<object>} create
+   * @param {function(acquisitionContext: object, address: ServerAddress, function(address: ServerAddress, resource: object): Promise<object>): Promise<object>} create
    *                an allocation function that creates a promise with a new resource. It's given an address for which to
    *                allocate the connection and a function that will return the resource to the pool if invoked, which is
    *                meant to be called on .dispose or .close or whatever mechanism the resource uses to finalize.
+   * @param {function(acquisitionContext: object, resource: object): boolean} validateOnAcquire
+   *                called at various times when an instance is acquired
+   *                If this returns false, the resource will be evicted
+   * @param {function(resource: object): boolean} validateOnRelease
+   *                called at various times when an instance is released
+   *                If this returns false, the resource will be evicted
    * @param {function(resource: object): Promise<void>} destroy
    *                called with the resource when it is evicted from this pool
-   * @param {function(resource: object): boolean} validate
-   *                called at various times (like when an instance is acquired and when it is returned.
-   *                If this returns false, the resource will be evicted
    * @param {function(resource: object, observer: { onError }): void} installIdleObserver
    *                called when the resource is released back to pool
    * @param {function(resource: object): void} removeIdleObserver
@@ -43,17 +46,19 @@ class Pool {
    * @param {Logger} log the driver logger.
    */
   constructor ({
-    create = (address, release) => Promise.resolve(),
+    create = (acquisitionContext, address, release) => Promise.resolve(),
     destroy = conn => Promise.resolve(),
-    validate = conn => true,
+    validateOnAcquire = (acquisitionContext, conn) => true,
+    validateOnRelease = (conn) => true,
     installIdleObserver = (conn, observer) => {},
     removeIdleObserver = conn => {},
     config = PoolConfig.defaultConfig(),
     log = Logger.noOp()
   } = {}) {
     this._create = create
     this._destroy = destroy
-    this._validate = validate
+    this._validateOnAcquire = validateOnAcquire
+    this._validateOnRelease = validateOnRelease
     this._installIdleObserver = installIdleObserver
     this._removeIdleObserver = removeIdleObserver
     this._maxSize = config.maxSize
@@ -69,10 +74,11 @@ class Pool {
 
   /**
    * Acquire and idle resource fom the pool or create a new one.
+   * @param {object} acquisitionContext the acquisition context used for create and validateOnAcquire connection
    * @param {ServerAddress} address the address for which we're acquiring.
    * @return {Promise<Object>} resource that is ready to use.
    */
-  acquire (address) {
+  acquire (acquisitionContext, address) {
     const key = address.asKey()
 
     // We're out of resources and will try to acquire later on when an existing resource is released.
@@ -108,7 +114,7 @@ class Pool {
         }
       }, this._acquisitionTimeout)
 
-      request = new PendingRequest(key, resolve, reject, timeoutId, this._log)
+      request = new PendingRequest(key, acquisitionContext, resolve, reject, timeoutId, this._log)
       allRequests[key].push(request)
       this._processPendingAcquireRequests(address)
     })
@@ -193,7 +199,7 @@ class Pool {
     return pool
   }
 
-  async _acquire (address) {
+  async _acquire (acquisitionContext, address) {
     if (this._closed) {
       throw newError('Pool is closed, it is no more able to serve requests.')
     }
@@ -207,7 +213,7 @@ class Pool {
         this._removeIdleObserver(resource)
       }
 
-      if (await this._validate(resource)) {
+      if (await this._validateOnAcquire(acquisitionContext, resource)) {
         // idle resource is valid and can be acquired
         resourceAcquired(key, this._activeResourceCounts)
         if (this._log.isDebugEnabled()) {
@@ -238,7 +244,7 @@ class Pool {
     let resource
     try {
       // Invoke callback that creates actual connection
-      resource = await this._create(address, (address, resource) => this._release(address, resource, pool))
+      resource = await this._create(acquisitionContext, address, (address, resource) => this._release(address, resource, pool))
 
       pool.pushInUse(resource)
       resourceAcquired(key, this._activeResourceCounts)
@@ -256,7 +262,7 @@ class Pool {
 
     if (pool.isActive()) {
       // there exist idle connections for the given key
-      if (!await this._validate(resource)) {
+      if (!await this._validateOnRelease(resource)) {
         if (this._log.isDebugEnabled()) {
           this._log.debug(
             `${resource} destroyed and can't be released to the pool ${key} because it is not functional`
@@ -327,7 +333,7 @@ class Pool {
       const pendingRequest = requests.shift() // pop a pending acquire request
 
       if (pendingRequest) {
-        this._acquire(address)
+        this._acquire(pendingRequest.context, address)
           .catch(error => {
             // failed to acquire/create a new connection to resolve the pending acquire request
             // propagate the error by failing the pending request
@@ -391,15 +397,20 @@ function resourceReleased (key, activeResourceCounts) {
 }
 
 class PendingRequest {
-  constructor (key, resolve, reject, timeoutId, log) {
+  constructor (key, context, resolve, reject, timeoutId, log) {
     this._key = key
+    this._context = context
     this._resolve = resolve
     this._reject = reject
     this._timeoutId = timeoutId
     this._log = log
     this._completed = false
   }
 
+  get context () {
+    return this._context
+  }
+
   isCompleted () {
     return this._completed
   }

packages/bolt-connection/test/connection-provider/connection-provider-direct.test.js

@@ -333,7 +333,7 @@ function newPool ({ create, config } = {}) {
   }
   return new Pool({
     config,
-    create: (address, release) =>
+    create: (_, address, release) =>
       Promise.resolve(_create(address, release))
   })
 }

packages/bolt-connection/test/connection-provider/connection-provider-routing.test.js

@@ -129,9 +129,9 @@ describe.each([
   it('purges connections when address is forgotten', () => {
     const pool = newPool()
 
-    pool.acquire(server1)
-    pool.acquire(server3)
-    pool.acquire(server5)
+    pool.acquire({}, server1)
+    pool.acquire({}, server3)
+    pool.acquire({}, server5)
     expectPoolToContain(pool, [server1, server3, server5])
 
     const connectionProvider = newRoutingConnectionProvider(
@@ -2589,7 +2589,7 @@ describe.each([
 
         const targetServers = accessMode === WRITE ? routingTable.writers : routingTable.readers
         const address = targetServers[0]
-        expect(acquireSpy).toHaveBeenCalledWith(address)
+        expect(acquireSpy).toHaveBeenCalledWith({}, address)
 
         const connections = seenConnectionsPerAddress.get(address)
 
@@ -2608,7 +2608,7 @@ describe.each([
 
         const targetServers = accessMode === WRITE ? routingTable.writers : routingTable.readers
         const address = targetServers[0]
-        expect(acquireSpy).toHaveBeenCalledWith(address)
+        expect(acquireSpy).toHaveBeenCalledWith({}, address)
 
         const connections = seenConnectionsPerAddress.get(address)
 
@@ -2628,7 +2628,7 @@ describe.each([
 
         const targetServers = accessMode === WRITE ? routingTable.readers : routingTable.writers
         for (const address of targetServers) {
-          expect(acquireSpy).not.toHaveBeenCalledWith(address)
+          expect(acquireSpy).not.toHaveBeenCalledWith({}, address)
           expect(seenConnectionsPerAddress.get(address)).toBeUndefined()
         }
       })
@@ -2711,7 +2711,7 @@ describe.each([
           } finally {
             const targetServers = accessMode === WRITE ? routingTable.writers : routingTable.readers
             for (const address of targetServers) {
-              expect(acquireSpy).toHaveBeenCalledWith(address)
+              expect(acquireSpy).toHaveBeenCalledWith({}, address)
 
               const connections = seenConnectionsPerAddress.get(address)
 
@@ -2947,7 +2947,7 @@ describe.each([
     }
     return new Pool({
       config,
-      create: (address, release) => _create(address, release)
+      create: (_, address, release) => _create(address, release)
     })
   }