Correctly unpack failure message to Neo4jError

Previously raw FAILURE message was passed to `onError`-like callbacks and
thrown as is. This means thrown error contained packstream related fields and
actually was a packstream structure. Example:

```
Structure {
  signature: 127,
  fields: [
    {
      code: 'Neo.ClientError.Schema.ConstraintValidationFailed',
      message: 'Node 0 already exists with label A and property "a"=[42]'
    }
  ]
}
```

This commit makes sure we unpack/convert FAILURE messages into `Neo4jError`s
and pass those further to callbacks.

Author: lutovich

src/v1/internal/buf.js

@@ -22,7 +22,6 @@
   *(via Buffer API).
   */
 
-import {newError} from './../error';
 let _node = require("buffer");
 /**
   * Common base with default implementation for most buffer methods.
@@ -551,7 +550,7 @@ class NodeBuffer extends BaseBuffer {
         val.position + bytesToCopy );
       val.position += bytesToCopy;
     } else {
-      throw newError("Copying not yet implemented.");
+      super.putBytes(position, val);
     }
   };
 

src/v1/internal/connector.js

@@ -18,11 +18,11 @@
  */
 import WebSocketChannel from './ch-websocket';
 import NodeChannel from './ch-node';
-import {Dechunker, Chunker} from "./chunking";
+import {Chunker, Dechunker} from './chunking';
 import hasFeature from './features';
 import {Packer, Unpacker} from './packstream';
 import {alloc} from './buf';
-import {Node, Relationship, UnboundRelationship, Path, PathSegment} from '../graph-types'
+import {Node, Path, PathSegment, Relationship, UnboundRelationship} from '../graph-types';
 import {newError} from './../error';
 
 let Channel;
@@ -201,7 +201,9 @@ class Connection {
     this._chunker = new Chunker( channel );
     this._packer = new Packer( this._chunker );
     this._unpacker = new Unpacker();
+
     this._isHandlingFailure = false;
+    this._currentFailure = null;
 
     // Set to true on fatal errors, to get this out of session pool.
     this._isBroken = false;
@@ -288,54 +290,56 @@ class Connection {
   }
 
   _handleMessage( msg ) {
+    const payload = msg.fields[0];
+
     switch( msg.signature ) {
       case RECORD:
-        log("S", "RECORD", msg.fields[0]);
-        this._currentObserver.onNext( msg.fields[0] );
+        log("S", "RECORD", msg);
+        this._currentObserver.onNext( payload );
         break;
       case SUCCESS:
-        log("S", "SUCCESS", msg.fields[0]);
+        log("S", "SUCCESS", msg);
         try {
-          this._currentObserver.onCompleted( msg.fields[0] );
+          this._currentObserver.onCompleted( payload );
         } finally {
           this._currentObserver = this._pendingObservers.shift();
         }
         break;
       case FAILURE:
         log("S", "FAILURE", msg);
         try {
-          this._currentObserver.onError( msg );
-          this._errorMsg = msg;
+          this._currentFailure = newError(payload.message, payload.code);
+          this._currentObserver.onError( this._currentFailure );
         } finally {
           this._currentObserver = this._pendingObservers.shift();
           // Things are now broken. Pending observers will get FAILURE messages routed until
           // We are done handling this failure.
           if( !this._isHandlingFailure ) {
             this._isHandlingFailure = true;
-            let self = this;
 
             // isHandlingFailure was false, meaning this is the first failure message
             // we see from this failure. We may see several others, one for each message
             // we had "optimistically" already sent after whatever it was that failed.
             // We only want to and need to ACK the first one, which is why we are tracking
             // this _isHandlingFailure thing.
             this._ackFailure({
-               onNext: NO_OP,
-               onError: NO_OP,
-               onCompleted: () => {
-                  self._isHandlingFailure = false;
-               }
+              onNext: NO_OP,
+              onError: NO_OP,
+              onCompleted: () => {
+                this._isHandlingFailure = false;
+                this._currentFailure = null;
+              }
             });
           }
         }
         break;
       case IGNORED:
-        log("S", "IGNORED");
+        log("S", "IGNORED", msg);
         try {
-          if (this._errorMsg && this._currentObserver.onError)
-            this._currentObserver.onError(this._errorMsg);
+          if (this._currentFailure && this._currentObserver.onError)
+            this._currentObserver.onError(this._currentFailure);
           else if(this._currentObserver.onError)
-            this._currentObserver.onError(msg);
+            this._currentObserver.onError(payload);
         } finally {
           this._currentObserver = this._pendingObservers.shift();
         }

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

@@ -31,7 +31,7 @@ export default class GetServersUtil {
       session.close();
       return result.records;
     }).catch(error => {
-      if (this._isProcedureNotFoundError(error)) {
+      if (error.code === PROCEDURE_NOT_FOUND_CODE) {
         // 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);
@@ -92,16 +92,4 @@ export default class GetServersUtil {
         PROTOCOL_ERROR);
     }
   }
-
-  _isProcedureNotFoundError(error) {
-    let errorCode = error.code;
-    if (!errorCode) {
-      try {
-        errorCode = error.fields[0].code;
-      } catch (e) {
-        errorCode = 'UNKNOWN';
-      }
-    }
-    return errorCode === PROCEDURE_NOT_FOUND_CODE;
-  }
 }

src/v1/routing-driver.js

@@ -36,37 +36,17 @@ class RoutingDriver extends Driver {
   }
 
   _createSession(connectionPromise, cb) {
-    return new RoutingSession(connectionPromise, cb, (err, conn) => {
-      let code = err.code;
-      let msg = err.message;
-      if (!code) {
-        try {
-          code = err.fields[0].code;
-        } catch (e) {
-          code = 'UNKNOWN';
-        }
-      }
-      if (!msg) {
-        try {
-          msg = err.fields[0].message;
-        } catch (e) {
-          msg = 'Unknown failure occurred';
-        }
-      }
-      //just to simplify later error handling
-      err.code = code;
-      err.message = msg;
-
-      if (code === SERVICE_UNAVAILABLE || code === SESSION_EXPIRED) {
+    return new RoutingSession(connectionPromise, cb, (error, conn) => {
+      if (error.code === SERVICE_UNAVAILABLE || error.code === SESSION_EXPIRED) {
         if (conn) {
           this._forget(conn.url)
         } else {
           connectionPromise.then((conn) => {
             this._forget(conn.url);
           }).catch(() => {/*ignore*/});
         }
-        return err;
-      } else if (code === 'Neo.ClientError.Cluster.NotALeader') {
+        return error;
+      } else if (error.code === 'Neo.ClientError.Cluster.NotALeader') {
         let url = 'UNKNOWN';
         if (conn) {
           url = conn.url;
@@ -76,9 +56,9 @@ class RoutingDriver extends Driver {
             this._routingTable.forgetWriter(conn.url);
           }).catch(() => {/*ignore*/});
         }
-        return newError("No longer possible to write to server at " + url, SESSION_EXPIRED);
+        return newError('No longer possible to write to server at ' + url, SESSION_EXPIRED);
       } else {
-        return err;
+        return error;
       }
     });
   }

test/internal/connector.test.js

@@ -17,8 +17,12 @@
  * limitations under the License.
  */
 
-import * as DummyChannel from "../../src/v1/internal/ch-dummy";
-import {connect} from "../../src/v1/internal/connector";
+import * as DummyChannel from '../../src/v1/internal/ch-dummy';
+import {connect, Connection} from '../../src/v1/internal/connector';
+import {Packer} from '../../src/v1/internal/packstream';
+import {Chunker} from '../../src/v1/internal/chunking';
+import {alloc} from '../../src/v1/internal/buf';
+import {Neo4jError} from '../../src/v1/error';
 
 describe('connector', () => {
 
@@ -94,4 +98,48 @@ describe('connector', () => {
 
   });
 
+  it('should convert failure messages to errors', done => {
+    const channel = new DummyChannel.channel;
+    const connection = new Connection(channel, 'bolt://localhost');
+
+    const errorCode = 'Neo.ClientError.Schema.ConstraintValidationFailed';
+    const errorMessage = 'Node 0 already exists with label User and property "email"=[john@doe.com]';
+
+    connection._queueObserver({
+      onError: error => {
+        expectNeo4jError(error, errorCode, errorMessage);
+        done();
+      }
+    });
+
+    channel.onmessage(packedHandshakeMessage());
+    channel.onmessage(packedFailureMessage(errorCode, errorMessage));
+  });
+
+  function packedHandshakeMessage() {
+    const result = alloc(4);
+    result.putInt32(0, 1);
+    result.reset();
+    return result;
+  }
+
+  function packedFailureMessage(code, message) {
+    const channel = new DummyChannel.channel;
+    const chunker = new Chunker(channel);
+    const packer = new Packer(chunker);
+    packer.packStruct(0x7F, [packer.packable({code: code, message: message})]);
+    chunker.messageBoundary();
+    chunker.flush();
+    const data = channel.toBuffer();
+    const result = alloc(data.length);
+    result.putBytes(0, data);
+    return result;
+  }
+
+  function expectNeo4jError(error, expectedCode, expectedMessage) {
+    expect(() => {
+      throw error;
+    }).toThrow(new Neo4jError(expectedMessage, expectedCode));
+  }
+
 });

test/v1/driver.test.js

@@ -79,7 +79,7 @@ describe('driver', function() {
     // Expect
     driver.onError = function (err) {
       //the error message is different whether in browser or node
-      expect(err.fields[0].code).toEqual('Neo.ClientError.Security.Unauthorized');
+      expect(err.code).toEqual('Neo.ClientError.Security.Unauthorized');
       done();
     };
 

test/v1/examples.test.js

@@ -306,7 +306,7 @@ describe('examples', function() {
     // end::handle-cypher-error[]
 
     testResultPromise.then(function(loggedError){
-      expect(loggedError.fields[0].code).toBe( "Neo.ClientError.Statement.SyntaxError" );
+      expect(loggedError.code).toBe( 'Neo.ClientError.Statement.SyntaxError' );
       done();
     });
   });

test/v1/session.test.js

@@ -94,9 +94,9 @@ describe('session', function () {
   it('should call observers onError on error ', function (done) {
 
     // When & Then
-    session.run("RETURN 1 AS").subscribe({
+    session.run('RETURN 1 AS').subscribe({
       onError: function (error) {
-        expect(error.fields.length).toBe(1);
+        expect(error.code).toEqual('Neo.ClientError.Statement.SyntaxError');
         done();
       }
     });
@@ -139,9 +139,9 @@ describe('session', function () {
 
   it('should expose basic run/catch ', function (done) {
     // When & Then
-    session.run("RETURN 1 AS").catch(
+    session.run('RETURN 1 AS').catch(
       function (error) {
-        expect(error.fields.length).toBe(1);
+        expect(error.code).toEqual('Neo.ClientError.Statement.SyntaxError');
         done();
       }
     )

test/v1/tck/steps/authsteps.js

@@ -64,8 +64,8 @@ module.exports = function () {
   });
 
   this.Then(/^a `Protocol Error` is raised$/, function () {
-    var message = this.err.fields[0].message;
-    var code = this.err.fields[0].code;
+    var message = this.err.message;
+    var code = this.err.code;
 
     var expectedStartOfMessage = 'The client is unauthorized due to authentication failure.';
     var expectedCode = 'Neo.ClientError.Security.Unauthorized';

test/v1/tck/steps/erroreportingsteps.js

@@ -58,7 +58,7 @@ module.exports = function () {
 
   this.When(/^I run a non valid cypher statement$/, function (callback) {
     var self = this;
-    this.session.run("CRETE (:n)").then(function(result) {callback()}).catch(function(err) {self.error = err.fields[0]; callback()})
+    this.session.run("CRETE (:n)").then(function(result) {callback()}).catch(function(err) {self.error = err; callback()})
   });
 
   this.When(/^I set up a driver to an incorrect port$/, function (callback) {

test/v1/transaction.test.js

@@ -77,20 +77,20 @@ describe('transaction', () => {
 
   it('should handle failures with subscribe', done => {
     const tx = session.beginTransaction();
-    tx.run("THIS IS NOT CYPHER")
+    tx.run('THIS IS NOT CYPHER')
       .catch(error => {
-        expect(error.fields.length).toBe(1);
+        expect(error.code).toEqual('Neo.ClientError.Statement.SyntaxError');
         driver.close();
         done();
       });
   });
 
   it('should handle failures with catch', done => {
     const tx = session.beginTransaction();
-    tx.run("THIS IS NOT CYPHER")
+    tx.run('THIS IS NOT CYPHER')
       .subscribe({
         onError: error => {
-          expect(error.fields.length).toBe(1);
+          expect(error.code).toEqual('Neo.ClientError.Statement.SyntaxError');
           driver.close();
           done();
         }
@@ -297,7 +297,7 @@ describe('transaction', () => {
     const invalidBookmark = 'hi, this is an invalid bookmark';
     const tx = session.beginTransaction(invalidBookmark);
     tx.run('RETURN 1').catch(error => {
-      expect(error.fields[0].code).toBe('Neo.ClientError.Transaction.InvalidBookmark');
+      expect(error.code).toBe('Neo.ClientError.Transaction.InvalidBookmark');
       done();
     });
   });
@@ -317,7 +317,7 @@ describe('transaction', () => {
         const unreachableBookmark = session.lastBookmark() + "0";
         const tx2 = session.beginTransaction(unreachableBookmark);
         tx2.run('CREATE ()').catch(error => {
-          const message = error.fields[0].message;
+          const message = error.message;
           const expectedPrefix = message.indexOf('Database not up to the requested version') === 0;
           expect(expectedPrefix).toBeTruthy();
           done();
@@ -445,8 +445,7 @@ describe('transaction', () => {
   });
 
   function expectSyntaxError(error) {
-    const code = error.fields[0].code;
-    expect(code).toBe('Neo.ClientError.Statement.SyntaxError');
+    expect(error.code).toBe('Neo.ClientError.Statement.SyntaxError');
   }
 
   function expectValidLastBookmark(session) {