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/authentication-provider.js

@@ -17,6 +17,8 @@
  * limitations under the License.
  */
 
+import { object } from '../lang'
+
 /**
  * Class which provides Authorization for {@link Connection}
  */
@@ -28,7 +30,14 @@ export default class AuthenticationProvider {
     this._refreshObserver = undefined
   }
 
-  async authenticate ({ connection }) {
+  async authenticate ({ connection, auth }) {
+    if (auth != null) {
+      if (connection.authToken == null || (connection.supportsReAuth && !object.equals(connection.authToken, auth))) {
+        return await connection.connect(this._userAgent, auth)
+      }
+      return connection
+    }
+
     if (!this._authToken || this._isTokenExpired) {
       await this._getFreshAuthToken()
     }

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

@@ -24,6 +24,7 @@ import {
   ConnectionErrorHandler
 } from '../connection'
 import { internal, error } from 'neo4j-driver-core'
+import { object } from '../lang'
 
 const {
   constants: { BOLT_PROTOCOL_V3, BOLT_PROTOCOL_V4_0, BOLT_PROTOCOL_V4_4 }
@@ -49,15 +50,11 @@ 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({ auth }, this._address)
 
-    if (auth && auth !== connection.authToken) {
-      if (connection.supportsReAuth) {
-        await connection.connect(this._userAgent, auth)
-      } else {
-        await connection._release()
-        return await this._createStickyConnection({ address: this._address, auth })
-      }
+    if (auth && !object.equals(auth, connection.authToken)) {
+      await connection._release()
+      return await this._createStickyConnection({ address: this._address, auth })
     }
 
     return new DelegateConnection(connection, databaseSpecificErrorHandler)

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,13 +70,13 @@ 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 ({ auth }, address, release) {
     return this._createChannelConnection(address).then(connection => {
       connection._release = () => {
         return release(address, connection)
       }
       this._openConnections[connection.id] = connection
-      return this._authenticationProvider.authenticate({ connection })
+      return this._authenticationProvider.authenticate({ connection, auth })
         .catch(error => {
           // let's destroy this connection
           this._destroyConnection(connection)
@@ -85,12 +86,32 @@ export default class PooledConnectionProvider extends ConnectionProvider {
     })
   }
 
+  async _validateConnectionOnAcquire ({ auth }, conn) {
+    if (!this._validateConnection(conn)) {
+      return false
+    }
+
+    try {
+      await this._authenticationProvider.authenticate({ connection: conn, auth })
+      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,16 +122,20 @@ export default class PooledConnectionProvider extends ConnectionProvider {
       return false
     }
 
+    return true
+  }
+
+  async _createStickyConnection ({ address, auth }) {
+    const connection = this._createChannelConnection(address)
+    connection._release = () => this._destroyConnection(connection)
+    this._openConnections[connection.id] = connection
+
     try {
-      await this._authenticationProvider.authenticate({ connection: conn })
+      return await connection.connect(this._userAgent, auth)
     } catch (error) {
-      this._log.info(
-        `The connection ${conn.id} is not valid because of an error ${error.code} '${error.message}'`
-      )
-      return false
+      await this._destroyConnection()
+      throw error
     }
-
-    return true
   }
 
   /**
@@ -130,7 +155,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

@@ -28,6 +28,7 @@ import {
   ConnectionErrorHandler,
   DelegateConnection
 } from '../connection'
+import { object } from '../lang'
 
 const { SERVICE_UNAVAILABLE, SESSION_EXPIRED } = error
 const {
@@ -186,19 +187,11 @@ export default class RoutingConnectionProvider extends PooledConnectionProvider
     }
 
     try {
-      const connection = await this._acquireConnectionToServer(
-        address,
-        name,
-        routingTable
-      )
+      const connection = await this._connectionPool.acquire({ auth }, address)
 
       if (auth && auth !== connection.authToken) {
-        if (connection.supportsReAuth) {
-          await connection.connect(this._userAgent, auth)
-        } else {
-          await connection._release()
-          return await this._createStickyConnection({ address, auth })
-        }
+        await connection._release()
+        return await this._createStickyConnection({ address, auth })
       }
 
       return new DelegateConnection(connection, databaseSpecificErrorHandler)
@@ -316,10 +309,6 @@ export default class RoutingConnectionProvider extends PooledConnectionProvider
     })
   }
 
-  _acquireConnectionToServer (address, serverName, routingTable) {
-    return this._connectionPool.acquire(address)
-  }
-
   _freshRoutingTable ({ accessMode, database, bookmarks, impersonatedUser, onDatabaseNameResolved, auth } = {}) {
     const currentRoutingTable = this._routingTableRegistry.get(
       database,
@@ -544,33 +533,16 @@ export default class RoutingConnectionProvider extends PooledConnectionProvider
     )
   }
 
-  async _createStickyConnection ({ address, auth }) {
-    const connection = await this._createChannelConnection(address)
-    connection._release = () => this._destroyConnection(connection)
-    this._openConnections[connection.id] = connection
-
-    try {
-      return await connection.connect(this._userAgent, auth)
-    } catch (error) {
-      await this._destroyConnection()
-      throw error
-    }
-  }
-
   async _createSessionForRediscovery (routerAddress, bookmarks, impersonatedUser, auth) {
     try {
-      let connection = await this._connectionPool.acquire(routerAddress)
+      let connection = await this._connectionPool.acquire({ auth }, routerAddress)
 
-      if (auth && connection.authToken !== auth) {
-        if (connection.supportsReAuth) {
-          await await connection.connect(this._userAgent, auth)
-        } else {
-          await connection._release()
-          connection = await this._createStickyConnection({
-            address: routerAddress,
-            auth
-          })
-        }
+      if (auth && object.equals(auth, connection.authToken)) {
+        await connection._release()
+        connection = await this._createStickyConnection({
+          address: routerAddress,
+          auth
+        })
       }
 
       const databaseSpecificErrorHandler = ConnectionErrorHandler.create({
@@ -767,7 +739,6 @@ function _isFailFastError (error) {
 }
 
 function _isFailFastSecurityError (error) {
-  console.error(error)
   return error.code.startsWith('Neo.ClientError.Security.') &&
     ![
       AUTHORIZATION_EXPIRED_CODE

packages/bolt-connection/src/lang/index.js

@@ -18,3 +18,4 @@
  */
 
 export * as functional from './functional'
+export * as object from './object'

packages/bolt-connection/src/lang/object.js

@@ -0,0 +1,43 @@
+/**
+ * Copyright (c) "Neo4j"
+ * Neo4j Sweden AB [http://neo4j.com]
+ *
+ * This file is part of Neo4j.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+export function equals (a, b) {
+  if (a === b) {
+    return true
+  }
+
+  if (typeof a === 'object' && typeof b === 'object') {
+    const keysA = Object.keys(a)
+    const keysB = Object.keys(b)
+
+    if (keysA.length !== keysB.length) {
+      return false
+    }
+
+    for (const key of keysA) {
+      if (a[key] !== b[key]) {
+        return false
+      }
+    }
+
+    return true
+  }
+
+  return false
+}