Replace ACK_FAILURE with RESET

Both have equivalent semantics with regards to failure handing.
ACK_FAILURE will also be removed in the next Bolt protocol version.

Author: lutovich

src/v1/internal/connector.js

@@ -44,7 +44,7 @@ else {
 let
 // Signature bytes for each message type
 INIT = 0x01,            // 0000 0001 // INIT <user_agent>
-ACK_FAILURE = 0x0E,     // 0000 1110 // ACK_FAILURE
+ACK_FAILURE = 0x0E,     // 0000 1110 // ACK_FAILURE - unused
 RESET = 0x0F,           // 0000 1111 // RESET
 RUN = 0x10,             // 0001 0000 // RUN <statement> <parameters>
 DISCARD_ALL = 0x2F,     // 0010 1111 // DISCARD *
@@ -106,7 +106,6 @@ class Connection {
     this._packer = packStreamUtil.createLatestPacker(this._chunker);
     this._unpacker = packStreamUtil.createLatestUnpacker(disableLosslessIntegers);
 
-    this._ackFailureMuted = false;
     this._currentFailure = null;
 
     this._state = new ConnectionState(this);
@@ -247,7 +246,7 @@ class Connection {
         } finally {
           this._updateCurrentObserver();
           // Things are now broken. Pending observers will get FAILURE messages routed until we are done handling this failure.
-          this._ackFailureIfNeeded();
+          this._resetOnFailure();
         }
         break;
       case IGNORED:
@@ -326,13 +325,8 @@ class Connection {
    * @return {Promise<void>} promise resolved when SUCCESS-message response arrives, or failed when other response messages arrives.
    */
   resetAndFlush() {
-    if (this._log.isDebugEnabled()) {
-      this._log.debug(`${this} C: RESET`);
-    }
-    this._ackFailureMuted = true;
-
     return new Promise((resolve, reject) => {
-      const observer = {
+      this._reset({
         onNext: record => {
           const neo4jError = this._handleProtocolError('Received RECORD as a response for RESET: ' + JSON.stringify(record));
           reject(neo4jError);
@@ -347,48 +341,35 @@ class Connection {
           }
         },
         onCompleted: () => {
-          this._ackFailureMuted = false;
           resolve();
         }
-      };
-      const queued = this._queueObserver(observer);
-      if (queued) {
-        this._packer.packStruct(RESET, [], err => this._handleFatalError(err));
-        this._chunker.messageBoundary();
-        this.sync();
-      }
+      });
     });
   }
 
-  _ackFailureIfNeeded() {
-    if (this._ackFailureMuted) {
-      return;
-    }
-
-    if (this._log.isDebugEnabled()) {
-      this._log.debug(`${this} C: ACK_FAILURE`);
-    }
-
-    const observer = {
+  _resetOnFailure() {
+    this._reset({
       onNext: record => {
-        this._handleProtocolError('Received RECORD as a response for ACK_FAILURE: ' + JSON.stringify(record));
+        this._handleProtocolError('Received RECORD as a response for RESET: ' + JSON.stringify(record));
       },
-      onError: error => {
-        if (!this._isBroken && !this._ackFailureMuted) {
-          // not handling a fatal error and RESET did not cause the given error - looks like a protocol violation
-          this._handleProtocolError('Received FAILURE as a response for ACK_FAILURE: ' + error);
-        } else {
-          this._currentFailure = null;
-        }
+      // clear the current failure when response for RESET is received
+      onError: () => {
+        this._currentFailure = null;
       },
       onCompleted: () => {
         this._currentFailure = null;
       }
-    };
+    });
+  }
+
+  _reset(observer) {
+    if (this._log.isDebugEnabled()) {
+      this._log.debug(`${this} C: RESET`);
+    }
 
     const queued = this._queueObserver(observer);
     if (queued) {
-      this._packer.packStruct(ACK_FAILURE, [], err => this._handleFatalError(err));
+      this._packer.packStruct(RESET, [], err => this._handleFatalError(err));
       this._chunker.messageBoundary();
       this.sync();
     }
@@ -477,7 +458,6 @@ class Connection {
   }
 
   _handleProtocolError(message) {
-    this._ackFailureMuted = false;
     this._currentFailure = null;
     this._updateCurrentObserver();
     const error = newError(message, PROTOCOL_ERROR);

test/internal/connector.test.js

@@ -270,10 +270,6 @@ describe('connector', () => {
     testQueueingOfObserversWithBrokenConnection(resetAction);
   });
 
-  it('should not queue ACK_FAILURE observer when broken', () => {
-    testQueueingOfObserversWithBrokenConnection(connection => connection._ackFailureIfNeeded());
-  });
-
   it('should reset and flush when SUCCESS received', done => {
     connection = connect('bolt://localhost');
 
@@ -315,36 +311,16 @@ describe('connector', () => {
     connection._handleMessage(RECORD_MESSAGE);
   });
 
-  it('should ACK_FAILURE when SUCCESS received', () => {
+  it('should acknowledge failure with RESET when SUCCESS received', () => {
     connection = connect('bolt://localhost');
 
     connection._currentFailure = newError('Hello');
-    connection._ackFailureIfNeeded();
+    connection._resetOnFailure();
 
     connection._handleMessage(SUCCESS_MESSAGE);
     expect(connection._currentFailure).toBeNull();
   });
 
-  it('should fail the connection when ACK_FAILURE receives FAILURE', () => {
-    connection = connect('bolt://localhost');
-
-    connection._ackFailureIfNeeded();
-
-    connection._handleMessage(FAILURE_MESSAGE);
-    expect(connection._isBroken).toBeTruthy();
-    expect(connection.isOpen()).toBeFalsy();
-  });
-
-  it('should fail the connection when ACK_FAILURE receives RECORD', () => {
-    connection = connect('bolt://localhost');
-
-    connection._ackFailureIfNeeded();
-
-    connection._handleMessage(RECORD_MESSAGE);
-    expect(connection._isBroken).toBeTruthy();
-    expect(connection.isOpen()).toBeFalsy();
-  });
-
   function packedHandshakeMessage() {
     const result = alloc(4);
     result.putInt32(0, 1);

test/resources/boltstub/address_unavailable_template.script.mst

@@ -1,7 +1,6 @@
 !: AUTO INIT
 !: AUTO RESET
 !: AUTO PULL_ALL
-!: AUTO ACK_FAILURE
 !: AUTO RUN "ROLLBACK" {}
 !: AUTO RUN "BEGIN" {}
 !: AUTO RUN "COMMIT" {}

test/resources/boltstub/query_with_error.script

@@ -0,0 +1,11 @@
+!: AUTO INIT
+!: AUTO PULL_ALL
+
+C: RUN "RETURN 10 / 0" {}
+C: PULL_ALL
+S: FAILURE {"code": "Neo.ClientError.Statement.ArithmeticError", "message": "/ by zero"}
+S: IGNORED
+C: RESET
+S: SUCCESS
+C: RESET
+S: SUCCESS

test/v1/direct.driver.boltkit.test.js

@@ -304,4 +304,33 @@ describe('direct driver with stub server', () => {
     });
   });
 
+  it('should send RESET on error', done => {
+    if (!boltStub.supported) {
+      done();
+      return;
+    }
+
+    const server = boltStub.start('./test/resources/boltstub/query_with_error.script', 9001);
+
+    boltStub.run(() => {
+      const driver = boltStub.newDriver('bolt://127.0.0.1:9001');
+      const session = driver.session();
+
+      session.run('RETURN 10 / 0').then(result => {
+        done.fail('Should fail but received a result: ' + JSON.stringify(result));
+      }).catch(error => {
+        expect(error.code).toEqual('Neo.ClientError.Statement.ArithmeticError');
+        expect(error.message).toEqual('/ by zero');
+
+        session.close(() => {
+          driver.close();
+          server.exit(code => {
+            expect(code).toEqual(0);
+            done();
+          });
+        });
+      });
+    });
+  });
+
 });