Improve ordering of connection initialization

Make sure every connection acquired from the connection pool is fully
initialized and authenticated. Before this change, a connection was
created in a synchronous way. Then protocol handshake and INIT message
were sent asynchronously. So creation of a connection looked like a
sync operation but it is not. Authentication info was also sent with
INIT message regardless of the Bolt protocol negotiation. This worked
fine for Bolt V1 and V2 because they used same INIT message. However,
for Bolt V3 initialization has to happen after protocol version
negotiation because the initialization message is different.

Commit also moves error handling to the connection layer. Before, error
handling/processing was in both `StreamObserver` and `Transaction`.
It's mostly needed to handle failures in RoutingDriver (forget address
from the routing table). Now handling of errors will be done in a
single place - `Connection` object using a special `ConnectionErrorHandler`.
This handler is different for direct driver and routing drivers.

Author: lutovich

src/v1/driver.js

@@ -19,14 +19,15 @@
 
 import Session from './session';
 import Pool from './internal/pool';
-import {connect} from './internal/connector';
+import Connection from './internal/connection';
 import StreamObserver from './internal/stream-observer';
 import {newError, SERVICE_UNAVAILABLE} from './error';
 import {DirectConnectionProvider} from './internal/connection-providers';
 import Bookmark from './internal/bookmark';
 import ConnectivityVerifier from './internal/connectivity-verifier';
 import PoolConfig, {DEFAULT_ACQUISITION_TIMEOUT, DEFAULT_MAX_SIZE} from './internal/pool-config';
 import Logger from './internal/logger';
+import ConnectionErrorHandler from './internal/connection-error-handler';
 
 const DEFAULT_MAX_CONNECTION_LIFETIME = 60 * 60 * 1000; // 1 hour
 
@@ -62,18 +63,18 @@ class Driver {
    * @constructor
    * @param {string} hostPort
    * @param {string} userAgent
-   * @param {object} token
+   * @param {object} authToken
    * @param {object} config
    * @protected
    */
-  constructor(hostPort, userAgent, token = {}, config = {}) {
+  constructor(hostPort, userAgent, authToken = {}, config = {}) {
     sanitizeConfig(config);
 
     this._id = idGenerator++;
     this._hostPort = hostPort;
     this._userAgent = userAgent;
     this._openConnections = {};
-    this._token = token;
+    this._authToken = authToken;
     this._config = config;
     this._log = Logger.create(config);
     this._pool = new Pool(
@@ -127,18 +128,24 @@ class Driver {
   }
 
   /**
-   * Create a new connection instance.
-   * @return {Connection} new connector-api session instance, a low level session API.
+   * Create a new connection and initialize it.
+   * @return {Promise<Connection>} promise resolved with a new connection or rejected when failed to connect.
    * @access private
    */
   _createConnection(hostPort, release) {
-    const conn = connect(hostPort, this._config, this._connectionErrorCode(), this._log);
-    const streamObserver = new _ConnectionStreamObserver(this, conn);
-    conn.protocol().initialize(this._userAgent, this._token, streamObserver);
-    conn._release = () => release(hostPort, conn);
-
-    this._openConnections[conn.id] = conn;
-    return conn;
+    const connection = Connection.create(hostPort, this._config, this._createConnectionErrorHandler(), this._log);
+    connection._release = () => release(hostPort, connection);
+    this._openConnections[connection.id] = connection;
+
+    return connection.connect(this._userAgent, this._authToken)
+      .catch(error => {
+        if (this.onError) {
+          // notify Driver.onError callback about connection initialization errors
+          this.onError(error);
+        }
+        // propagate the error because connection failed to connect / initialize
+        throw error;
+      });
   }
 
   /**
@@ -186,7 +193,7 @@ class Driver {
     const sessionMode = Driver._validateSessionMode(mode);
     const connectionProvider = this._getOrCreateConnectionProvider();
     const bookmark = new Bookmark(bookmarkOrBookmarks);
-    return this._createSession(sessionMode, connectionProvider, bookmark, this._config);
+    return new Session(sessionMode, connectionProvider, bookmark, this._config);
   }
 
   static _validateSessionMode(rawMode) {
@@ -203,14 +210,8 @@ class Driver {
   }
 
   // Extension point
-  _createSession(mode, connectionProvider, bookmark, config) {
-    return new Session(mode, connectionProvider, bookmark, config);
-  }
-
-  // Extension point
-  _connectionErrorCode() {
-    // connection errors might result in different error codes depending on the driver
-    return SERVICE_UNAVAILABLE;
+  _createConnectionErrorHandler() {
+    return new ConnectionErrorHandler(SERVICE_UNAVAILABLE);
   }
 
   _getOrCreateConnectionProvider() {

src/v1/internal/connection-error-handler.js

@@ -0,0 +1,74 @@
+/**
+ * Copyright (c) 2002-2018 "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.
+ */
+
+import {SERVICE_UNAVAILABLE, SESSION_EXPIRED} from '../error';
+
+export default class ConnectionErrorHandler {
+
+  constructor(errorCode, handleUnavailability, handleWriteFailure) {
+    this._errorCode = errorCode;
+    this._handleUnavailability = handleUnavailability || noOpHandler;
+    this._handleWriteFailure = handleWriteFailure || noOpHandler;
+  }
+
+  /**
+   * Error code to use for network errors.
+   * @return {string} the error code.
+   */
+  errorCode() {
+    return this._errorCode;
+  }
+
+  /**
+   * Handle and transform the error.
+   * @param {Neo4jError} error the original error.
+   * @param {string} hostPort the host and port of the connection where the error happened.
+   * @return {Neo4jError} new error that should be propagated to the user.
+   */
+  handleAndTransformError(error, hostPort) {
+    if (isAvailabilityError(error)) {
+      return this._handleUnavailability(error, hostPort);
+    }
+    if (isFailureToWrite(error)) {
+      return this._handleWriteFailure(error, hostPort);
+    }
+    return error;
+  }
+}
+
+function isAvailabilityError(error) {
+  if (error) {
+    return error.code === SESSION_EXPIRED ||
+      error.code === SERVICE_UNAVAILABLE ||
+      error.code === 'Neo.TransientError.General.DatabaseUnavailable';
+  }
+  return false;
+}
+
+function isFailureToWrite(error) {
+  if (error) {
+    return error.code === 'Neo.ClientError.Cluster.NotALeader' ||
+      error.code === 'Neo.ClientError.General.ForbiddenOnReadOnlyDatabase';
+  }
+  return false;
+}
+
+function noOpHandler(error) {
+  return error;
+}

src/v1/internal/connection-holder.js

@@ -55,7 +55,7 @@ export default class ConnectionHolder {
   getConnection(streamObserver) {
     return this._connectionPromise.then(connection => {
       streamObserver.resolveConnection(connection);
-      return connection.initializationCompleted();
+      return connection;
     });
   }
 

src/v1/internal/connection-providers.js

@@ -26,6 +26,8 @@ import hasFeature from './features';
 import {DnsHostNameResolver, DummyHostNameResolver} from './host-name-resolvers';
 import RoutingUtil from './routing-util';
 
+const UNAUTHORIZED_ERROR_CODE = 'Neo.ClientError.Security.Unauthorized';
+
 class ConnectionProvider {
 
   acquireConnection(mode) {
@@ -195,20 +197,32 @@ export class LoadBalancer extends ConnectionProvider {
 
         // try next router
         return this._createSessionForRediscovery(currentRouter).then(session => {
-          return this._rediscovery.lookupRoutingTableOnRouter(session, currentRouter)
+          if (session) {
+            return this._rediscovery.lookupRoutingTableOnRouter(session, currentRouter);
+          } else {
+            // unable to acquire connection and create session towards the current router
+            // return null to signal that the next router should be tried
+            return null;
+          }
         });
       });
     }, Promise.resolve(null));
   }
 
   _createSessionForRediscovery(routerAddress) {
-    return this._connectionPool.acquire(routerAddress).then(connection => {
-      // initialized connection is required for routing procedure call
-      // server version needs to be known to decide which routing procedure to use
-      const initializedConnectionPromise = connection.initializationCompleted();
-      const connectionProvider = new SingleConnectionProvider(initializedConnectionPromise);
-      return new Session(READ, connectionProvider);
-    });
+    return this._connectionPool.acquire(routerAddress)
+      .then(connection => {
+        const connectionProvider = new SingleConnectionProvider(connection);
+        return new Session(READ, connectionProvider);
+      })
+      .catch(error => {
+        // unable to acquire connection towards the given router
+        if (error && error.code === UNAUTHORIZED_ERROR_CODE) {
+          // auth error is a sign of a configuration issue, rediscovery should not proceed
+          throw error;
+        }
+        return null;
+      });
   }
 
   _applyRoutingTableIfPossible(newRoutingTable) {
@@ -257,14 +271,14 @@ export class LoadBalancer extends ConnectionProvider {
 
 export class SingleConnectionProvider extends ConnectionProvider {
 
-  constructor(connectionPromise) {
+  constructor(connection) {
     super();
-    this._connectionPromise = connectionPromise;
+    this._connection = connection;
   }
 
   acquireConnection(mode) {
-    const connectionPromise = this._connectionPromise;
-    this._connectionPromise = null;
-    return connectionPromise;
+    const connection = this._connection;
+    this._connection = null;
+    return Promise.resolve(connection);
   }
 }