Avoid forgetting address twice on error in routing driver

Routing driver removes address from the routing table on error. This
removal is triggered by a `StreamObserver` that observes the incoming
messages. Same observer is used for RUN and PULL_ALL. It used to
trigger the forget operation trice on every error. This wasn't harmful
but looked weird in logs.

This commit fixes the problem by making `StreamObserver` invoke the
provided callback only once.

Author: lutovich

src/v1/internal/stream-observer.js

@@ -110,11 +110,13 @@ class StreamObserver {
    * @param {Object} error - An error object
    */
   onError(error) {
-    const transformedError = this._errorTransformer(error, this._conn);
     if(this._hasFailed) {
       return;
     }
     this._hasFailed = true;
+
+    const transformedError = this._errorTransformer(error, this._conn);
+
     if( this._observer ) {
       if( this._observer.onError ) {
         this._observer.onError( transformedError );

src/v1/transaction.js

@@ -103,21 +103,14 @@ class Transaction {
   }
 
   _onError() {
-    if (this.isOpen()) {
-      // attempt to rollback, useful when Transaction#run() failed
-      return this.rollback().catch(ignoredError => {
-        // ignore all errors because it is best effort and transaction might already be rolled back
-      }).then(() => {
-        // after rollback attempt change this transaction's state to FAILED
-        this._state = _states.FAILED;
-      });
-    } else {
-      // error happened in in-active transaction, just to the cleanup and change state to FAILED
-      this._state = _states.FAILED;
-      this._onClose();
-      // no async actions needed - return resolved promise
-      return Promise.resolve();
-    }
+    // error will be "acknowledged" by sending a RESET message
+    // database will then forget about this transaction and cleanup all corresponding resources
+    // it is thus safe to move this transaction to a FAILED state and disallow any further interactions with it
+    this._state = _states.FAILED;
+    this._onClose();
+
+    // release connection back to the pool
+    return this._connectionHolder.releaseConnection();
   }
 }
 
@@ -126,15 +119,12 @@ class _TransactionStreamObserver extends StreamObserver {
   constructor(tx) {
     super(tx._errorTransformer || ((err) => {return err}));
     this._tx = tx;
-    //this is to to avoid multiple calls to onError caused by IGNORED
-    this._hasFailed = false;
   }
 
   onError(error) {
     if (!this._hasFailed) {
       this._tx._onError().then(() => {
         super.onError(error);
-        this._hasFailed = true;
       });
     }
   }

test/internal/stream-observer.test.js

@@ -142,6 +142,19 @@ describe('StreamObserver', () => {
     expect(receivedMetaData).toEqual({metaDataField1: 'value1', metaDataField2: 'value2'});
   });
 
+  it('invokes error transformer only once on error', () => {
+    const errors = [];
+    const streamObserver = new StreamObserver(error => errors.push(error));
+
+    const error1 = new Error('Hello');
+    const error2 = new Error('World');
+
+    streamObserver.onError(error1);
+    streamObserver.onError(error2);
+
+    expect(errors).toEqual([error1]);
+  });
+
 });
 
 function newStreamObserver() {