Merge branch 1.2 into 1.3

Author: lutovich

src/v1/internal/connector.js

@@ -254,7 +254,7 @@ class Connection {
    * failing, and the connection getting ejected from the session pool.
    *
    * @param err an error object, forwarded to all current and future subscribers
-   * @private
+   * @protected
    */
   _handleFatalError( err ) {
     this._isBroken = true;
@@ -271,6 +271,12 @@ class Connection {
   }
 
   _handleMessage( msg ) {
+    if (this._isBroken) {
+      // ignore all incoming messages when this connection is broken. all previously pending observers failed
+      // with the fatal error. all future observers will fail with same fatal error.
+      return;
+    }
+
     const payload = msg.fields[0];
 
     switch( msg.signature ) {
@@ -283,7 +289,7 @@ class Connection {
         try {
           this._currentObserver.onCompleted( payload );
         } finally {
-          this._currentObserver = this._pendingObservers.shift();
+          this._updateCurrentObserver();
         }
         break;
       case FAILURE:
@@ -292,7 +298,7 @@ class Connection {
           this._currentFailure = newError(payload.message, payload.code);
           this._currentObserver.onError( this._currentFailure );
         } finally {
-          this._currentObserver = this._pendingObservers.shift();
+          this._updateCurrentObserver();
           // Things are now broken. Pending observers will get FAILURE messages routed until
           // We are done handling this failure.
           if( !this._isHandlingFailure ) {
@@ -322,7 +328,7 @@ class Connection {
           else if(this._currentObserver.onError)
             this._currentObserver.onError(payload);
         } finally {
-          this._currentObserver = this._pendingObservers.shift();
+          this._updateCurrentObserver();
         }
         break;
       default:
@@ -429,6 +435,14 @@ class Connection {
     return this._state.initializationCompleted();
   }
 
+  /*
+   * Pop next pending observer form the list of observers and make it current observer.
+   * @protected
+   */
+  _updateCurrentObserver() {
+    this._currentObserver = this._pendingObservers.shift();
+  }
+
   /**
    * Synchronize - flush all queued outgoing messages and route their responses
    * to their respective handlers.
@@ -480,7 +494,8 @@ class ConnectionState {
   }
 
   /**
-   * Wrap the given observer to track connection's initialization state.
+   * Wrap the given observer to track connection's initialization state. Connection is closed by the server if
+   * processing of INIT message fails so returned observer will handle initialization failure as a fatal error.
    * @param {StreamObserver} observer the observer used for INIT message.
    * @return {StreamObserver} updated observer.
    */
@@ -497,8 +512,14 @@ class ConnectionState {
           this._rejectPromise(error);
           this._rejectPromise = null;
         }
-        if (observer && observer.onError) {
-          observer.onError(error);
+
+        this._connection._updateCurrentObserver(); // make sure this same observer will not be called again
+        try {
+          if (observer && observer.onError) {
+            observer.onError(error);
+          }
+        } finally {
+          this._connection._handleFatalError(error);
         }
       },
       onCompleted: metaData => {

src/v1/internal/routing-util.js

@@ -26,6 +26,7 @@ const CALL_GET_SERVERS = 'CALL dbms.cluster.routing.getServers';
 const GET_ROUTING_TABLE_PARAM = 'context';
 const CALL_GET_ROUTING_TABLE = 'CALL dbms.cluster.routing.getRoutingTable({' + GET_ROUTING_TABLE_PARAM + '})';
 const PROCEDURE_NOT_FOUND_CODE = 'Neo.ClientError.Procedure.ProcedureNotFound';
+const UNAUTHORIZED_CODE = 'Neo.ClientError.Security.Unauthorized';
 
 export default class RoutingUtil {
 
@@ -49,10 +50,14 @@ export default class RoutingUtil {
         // throw when getServers procedure not found because this is clearly a configuration issue
         throw newError('Server ' + routerAddress + ' could not perform routing. ' +
           'Make sure you are connecting to a causal cluster', SERVICE_UNAVAILABLE);
+      } else if (error.code === UNAUTHORIZED_CODE) {
+        // auth error is a sign of a configuration issue, rediscovery should not proceed
+        throw error;
+      } else {
+        // return nothing when failed to connect because code higher in the callstack is still able to retry with a
+        // different session towards a different router
+        return null;
       }
-      // return nothing when failed to connect because code higher in the callstack is still able to retry with a
-      // different session towards a different router
-      return null;
     });
   }
 

test/internal/connector.test.js

@@ -145,6 +145,7 @@ describe('connector', () => {
 
     connection.initialize('mydriver/0.0.0', basicAuthToken(), {
       onCompleted: metaData => {
+        expect(connection.isOpen()).toBeTruthy();
         expect(metaData).toBeDefined();
         done();
       },
@@ -156,6 +157,7 @@ describe('connector', () => {
 
     connection.initialize('mydriver/0.0.0', basicAuthToken(), {
       onError: error => {
+        expect(connection.isOpen()).toBeFalsy();
         expect(error).toBeDefined();
         done();
       },
@@ -174,6 +176,30 @@ describe('connector', () => {
     connection.initialize('mydriver/0.0.0', basicAuthToken());
   });
 
+  it('should fail all new observers after initialization error', done => {
+    const connection = connect('bolt://localhost:7474'); // wrong port
+
+    connection.initialize('mydriver/0.0.0', basicAuthToken(), {
+      onError: initialError => {
+        expect(initialError).toBeDefined();
+
+        connection.run('RETURN 1', {}, {
+          onError: error1 => {
+            expect(error1).toEqual(initialError);
+
+            connection.initialize('mydriver/0.0.0', basicAuthToken(), {
+              onError: error2 => {
+                expect(error2).toEqual(initialError);
+
+                done();
+              }
+            });
+          }
+        });
+      },
+    });
+  });
+
   function packedHandshakeMessage() {
     const result = alloc(4);
     result.putInt32(0, 1);

test/resources/boltkit/failed_auth.script

@@ -0,0 +1,6 @@
+!: AUTO RESET
+!: AUTO PULL_ALL
+
+C: INIT "neo4j-javascript/0.0.0-dev" {"credentials": "neo4j", "scheme": "basic", "principal": "neo4j"}
+S: FAILURE {"code": "Neo.ClientError.Security.Unauthorized", "message": "Some server auth error message"}
+S: <EXIT>

test/v1/driver.test.js

@@ -80,7 +80,7 @@ describe('driver', () => {
 
   it('should fail early on wrong credentials', done => {
     // Given
-    driver = neo4j.driver("bolt://localhost", neo4j.auth.basic("neo4j", "who would use such a password"));
+    driver = neo4j.driver("bolt://localhost", wrongCredentials());
 
     // Expect
     driver.onError = err => {
@@ -93,6 +93,16 @@ describe('driver', () => {
     startNewTransaction(driver);
   });
 
+  it('should fail queries on wrong credentials', done => {
+    driver = neo4j.driver('bolt://localhost', wrongCredentials());
+
+    const session = driver.session();
+    session.run('RETURN 1').catch(error => {
+      expect(error.code).toEqual('Neo.ClientError.Security.Unauthorized');
+      done();
+    });
+  });
+
   it('should indicate success early on correct credentials', done => {
     // Given
     driver = neo4j.driver("bolt://localhost", sharedNeo4j.authToken);
@@ -211,4 +221,8 @@ describe('driver', () => {
     expect(session.beginTransaction()).toBeDefined();
   }
 
+  function wrongCredentials() {
+    return neo4j.auth.basic('neo4j', 'who would use such a password');
+  }
+
 });

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

@@ -1831,6 +1831,33 @@ describe('routing driver', () => {
     });
   });
 
+  it('should fail rediscovery on auth error', done => {
+    if (!boltkit.BoltKitSupport) {
+      done();
+      return;
+    }
+
+    const kit = new boltkit.BoltKit();
+    const router = kit.start('./test/resources/boltkit/failed_auth.script', 9010);
+
+    kit.run(() => {
+      const driver = newDriver('bolt+routing://127.0.0.1:9010');
+      const session = driver.session();
+      session.run('RETURN 1').catch(error => {
+        expect(error.code).toEqual('Neo.ClientError.Security.Unauthorized');
+        expect(error.message).toEqual('Some server auth error message');
+
+        session.close(() => {
+          driver.close();
+          router.exit(code => {
+            expect(code).toEqual(0);
+            done();
+          });
+        });
+      });
+    });
+  });
+
   function moveNextDateNow30SecondsForward() {
     const currentTime = Date.now();
     hijackNextDateNowCall(currentTime + 30 * 1000 + 1);