Fix purging of connections on INIT failure

Commit moves waiting for connection initialization promise higher
in the stack - from `ConnectionProvider` to `ConnectionHolder`.
This is needed to be able to notify stream observer about used
connection even if initialization fails. It is later possible to
purge connections for the given server. Previously stream observer
did not get to know the connection and pool was not purged on
connectivity errors.

Author: lutovich

src/v1/internal/connection-holder.js

@@ -49,10 +49,14 @@ export default class ConnectionHolder {
 
   /**
    * Get the current connection promise.
+   * @param {StreamObserver} streamObserver an observer for this connection.
    * @return {Promise<Connection>} promise resolved with the current connection.
    */
-  getConnection() {
-    return this._connectionPromise;
+  getConnection(streamObserver) {
+    return this._connectionPromise.then(connection => {
+      streamObserver.resolveConnection(connection);
+      return connection.initializationCompleted();
+    });
   }
 
   /**
@@ -117,7 +121,7 @@ class EmptyConnectionHolder extends ConnectionHolder {
     // nothing to initialize
   }
 
-  getConnection() {
+  getConnection(streamObserver) {
     return Promise.reject(newError('This connection holder does not serve connections'));
   }
 

src/v1/internal/connection-providers.js

@@ -54,7 +54,8 @@ export class DirectConnectionProvider extends ConnectionProvider {
   }
 
   acquireConnection(mode) {
-    const connectionPromise = acquireConnectionFromPool(this._connectionPool, this._address);
+    const connection = this._connectionPool.acquire(this._address);
+    const connectionPromise = Promise.resolve(connection);
     return this._withAdditionalOnErrorCallback(connectionPromise, this._driverOnErrorCallback);
   }
 }
@@ -101,7 +102,7 @@ export class LoadBalancer extends ConnectionProvider {
         `Failed to obtain connection towards ${serverName} server. Known routing table is: ${this._routingTable}`,
         SESSION_EXPIRED));
     }
-    return acquireConnectionFromPool(this._connectionPool, address);
+    return this._connectionPool.acquire(address);
   }
 
   _freshRoutingTable(accessMode) {
@@ -198,9 +199,10 @@ export class LoadBalancer extends ConnectionProvider {
   }
 
   _createSessionForRediscovery(routerAddress) {
+    const connection = this._connectionPool.acquire(routerAddress);
     // initialized connection is required for routing procedure call
     // server version needs to be known to decide which routing procedure to use
-    const initializedConnectionPromise = acquireConnectionFromPool(this._connectionPool, routerAddress);
+    const initializedConnectionPromise = connection.initializationCompleted();
     const connectionProvider = new SingleConnectionProvider(initializedConnectionPromise);
     return new Session(READ, connectionProvider);
   }
@@ -261,19 +263,3 @@ export class SingleConnectionProvider extends ConnectionProvider {
     return connectionPromise;
   }
 }
-
-// todo: test that all connection providers return initialized connections
-
-/**
- * Acquire an initialized connection from the given connection pool for the given address. Returned connection
- * promise will be resolved by a connection which completed initialization, i.e. received a SUCCESS response
- * for it's INIT message.
- * @param {Pool} connectionPool the connection pool to acquire connection from.
- * @param {string} address the server address.
- * @return {Promise<Connection>} the initialized connection.
- */
-function acquireConnectionFromPool(connectionPool, address) {
-  const connection = connectionPool.acquire(address);
-  // initialized connection is required to be able to perform subsequent server version checks
-  return connection.initializationCompleted();
-}

src/v1/internal/connector.js

@@ -498,11 +498,15 @@ class ConnectionState {
   constructor(connection) {
     this._connection = connection;
 
-    this._initialized = false;
-    this._initializationError = null;
-
-    this._resolvePromise = null;
-    this._rejectPromise = null;
+    this._initRequested = false;
+    this._initError = null;
+
+    this._resolveInitPromise = null;
+    this._rejectInitPromise = null;
+    this._initPromise = new Promise((resolve, reject) => {
+      this._resolveInitPromise = resolve;
+      this._rejectInitPromise = reject;
+    });
   }
 
   /**
@@ -519,11 +523,7 @@ class ConnectionState {
         }
       },
       onError: error => {
-        this._initializationError = error;
-        if (this._rejectPromise) {
-          this._rejectPromise(error);
-          this._rejectPromise = null;
-        }
+        this._processFailure(error);
 
         this._connection._updateCurrentObserver(); // make sure this same observer will not be called again
         try {
@@ -536,11 +536,8 @@ class ConnectionState {
       },
       onCompleted: metaData => {
         this._connection._markInitialized(metaData);
-        this._initialized = true;
-        if (this._resolvePromise) {
-          this._resolvePromise(this._connection);
-          this._resolvePromise = null;
-        }
+        this._resolveInitPromise(this._connection);
+
         if (observer && observer.onCompleted) {
           observer.onCompleted(metaData);
         }
@@ -553,15 +550,28 @@ class ConnectionState {
    * @return {Promise<Connection>} the result of connection initialization.
    */
   initializationCompleted() {
-    if (this._initialized) {
-      return Promise.resolve(this._connection);
-    } else if (this._initializationError) {
-      return Promise.reject(this._initializationError);
+    this._initRequested = true;
+
+    if (this._initError) {
+      const error = this._initError;
+      this._initError = null; // to reject initPromise only once
+      this._rejectInitPromise(error);
+    }
+
+    return this._initPromise;
+  }
+
+  /**
+   * @private
+   */
+  _processFailure(error) {
+    if (this._initRequested) {
+      // someone is waiting for initialization to complete, reject the promise
+      this._rejectInitPromise(error);
     } else {
-      return new Promise((resolve, reject) => {
-        this._resolvePromise = resolve;
-        this._rejectPromise = reject;
-      });
+      // no one is waiting for initialization, memorize the error but do not reject the promise
+      // to avoid unnecessary unhandled promise rejection warnings
+      this._initError = error;
     }
   }
 }

src/v1/session.js

@@ -75,8 +75,7 @@ class Session {
     const connectionHolder = this._connectionHolderWithMode(this._mode);
     if (!this._hasTx) {
       connectionHolder.initializeConnection();
-      connectionHolder.getConnection().then(connection => {
-        streamObserver.resolveConnection(connection);
+      connectionHolder.getConnection(streamObserver).then(connection => {
         statementRunner(connection, streamObserver);
         connection.pullAll(streamObserver);
         connection.sync();

src/v1/transaction.js

@@ -43,8 +43,7 @@ class Transaction {
       params = {bookmark: bookmark};
     }
 
-    this._connectionHolder.getConnection().then(conn => {
-      streamObserver.resolveConnection(conn);
+    this._connectionHolder.getConnection(streamObserver).then(conn => {
       conn.run('BEGIN', params, streamObserver);
       conn.pullAll(streamObserver);
     }).catch(error => streamObserver.onError(error));
@@ -167,8 +166,7 @@ let _states = {
       return {result: _runPullAll("ROLLBACK", connectionHolder, observer), state: _states.ROLLED_BACK};
     },
     run: (connectionHolder, observer, statement, parameters) => {
-      connectionHolder.getConnection().then(conn => {
-        observer.resolveConnection(conn);
+      connectionHolder.getConnection(observer).then(conn => {
         conn.run(statement, parameters || {}, observer);
         conn.pullAll(observer);
         conn.sync();
@@ -246,13 +244,11 @@ let _states = {
 };
 
 function _runPullAll(msg, connectionHolder, observer) {
-  connectionHolder.getConnection().then(
-    conn => {
-      observer.resolveConnection(conn);
-      conn.run(msg, {}, observer);
-      conn.pullAll(observer);
-      conn.sync();
-    }).catch(error => observer.onError(error));
+  connectionHolder.getConnection(observer).then(conn => {
+    conn.run(msg, {}, observer);
+    conn.pullAll(observer);
+    conn.sync();
+  }).catch(error => observer.onError(error));
 
   // for commit & rollback we need result that uses real connection holder and notifies it when
   // connection is not needed and can be safely released to the pool

test/internal/connection-holder.test.js

@@ -21,11 +21,12 @@ import ConnectionHolder, {EMPTY_CONNECTION_HOLDER} from '../../src/v1/internal/c
 import {SingleConnectionProvider} from '../../src/v1/internal/connection-providers';
 import {READ} from '../../src/v1/driver';
 import FakeConnection from './fake-connection';
+import StreamObserver from '../../src/v1/internal/stream-observer';
 
 describe('EmptyConnectionHolder', () => {
 
   it('should return rejected promise instead of connection', done => {
-    EMPTY_CONNECTION_HOLDER.getConnection().catch(() => {
+    EMPTY_CONNECTION_HOLDER.getConnection(new StreamObserver()).catch(() => {
       done();
     });
   });
@@ -62,8 +63,40 @@ describe('ConnectionHolder', () => {
 
     connectionHolder.initializeConnection();
 
-    connectionHolder.getConnection().then(connection => {
-      expect(connection).toBe(connection);
+    connectionHolder.getConnection(new StreamObserver()).then(conn => {
+      expect(conn).toBe(connection);
+      verifyConnectionInitialized(conn);
+      done();
+    });
+  });
+
+  it('should make stream observer aware about connection when initialization successful', done => {
+    const connection = new FakeConnection().withServerVersion('Neo4j/9.9.9');
+    const connectionProvider = newSingleConnectionProvider(connection);
+    const connectionHolder = new ConnectionHolder(READ, connectionProvider);
+    const streamObserver = new StreamObserver();
+
+    connectionHolder.initializeConnection();
+
+    connectionHolder.getConnection(streamObserver).then(conn => {
+      verifyConnectionInitialized(conn);
+      verifyConnection(streamObserver, 'Neo4j/9.9.9');
+      done();
+    });
+  });
+
+  it('should make stream observer aware about connection when initialization fails', done => {
+    const connection = new FakeConnection().withServerVersion('Neo4j/7.7.7').withFailedInitialization(new Error('Oh!'));
+    const connectionProvider = newSingleConnectionProvider(connection);
+    const connectionHolder = new ConnectionHolder(READ, connectionProvider);
+    const streamObserver = new StreamObserver();
+
+    connectionHolder.initializeConnection();
+
+    connectionHolder.getConnection(streamObserver).catch(error => {
+      expect(error.message).toEqual('Oh!');
+      verifyConnectionInitialized(connection);
+      verifyConnection(streamObserver, 'Neo4j/7.7.7');
       done();
     });
   });
@@ -195,3 +228,16 @@ class RecordingConnectionProvider extends SingleConnectionProvider {
 function newSingleConnectionProvider(connection) {
   return new SingleConnectionProvider(Promise.resolve(connection));
 }
+
+function verifyConnectionInitialized(connection) {
+  expect(connection.initializationInvoked).toEqual(1);
+}
+
+function verifyConnection(streamObserver, expectedServerVersion) {
+  expect(streamObserver._conn).toBeDefined();
+  expect(streamObserver._conn).not.toBeNull();
+
+  // server version is taken from connection, verify it as well
+  const metadata = streamObserver.serverMetadata();
+  expect(metadata.server.version).toEqual(expectedServerVersion);
+}

test/internal/connection-providers.test.js

@@ -44,16 +44,6 @@ describe('DirectConnectionProvider', () => {
     });
   });
 
-  it('returns an initialized connection', done => {
-    const pool = newPool();
-    const connectionProvider = newDirectConnectionProvider('localhost:123', pool);
-
-    connectionProvider.acquireConnection(READ).then(connection => {
-      expect(connection.initialized).toBeTruthy();
-      done();
-    });
-  });
-
 });
 
 describe('LoadBalancer', () => {
@@ -1059,26 +1049,6 @@ describe('LoadBalancer', () => {
     });
   });
 
-  it('returns an initialized connection', done => {
-    const pool = newPool();
-    const loadBalancer = newLoadBalancer(
-      ['server-1', 'server-2'],
-      ['server-3', 'server-4'],
-      ['server-5', 'server-6'],
-      pool
-    );
-
-    loadBalancer.acquireConnection(READ).then(connection => {
-      expect(connection.initialized).toBeTruthy();
-
-      loadBalancer.acquireConnection(WRITE).then(connection => {
-        expect(connection.initialized).toBeTruthy();
-
-        done();
-      });
-    });
-  });
-
 });
 
 function newDirectConnectionProvider(address, pool) {
@@ -1156,15 +1126,13 @@ class FakeConnection {
   constructor(address, release) {
     this.address = address;
     this.release = release;
-    this.initialized = false;
   }
 
   static create(address, release) {
     return new FakeConnection(address, release);
   }
 
   initializationCompleted() {
-    this.initialized = true;
     return Promise.resolve(this);
   }
 }

test/internal/fake-connection.js

@@ -31,9 +31,12 @@ export default class FakeConnection {
     this.resetAsyncInvoked = 0;
     this.syncInvoked = 0;
     this.releaseInvoked = 0;
+    this.initializationInvoked = 0;
     this.seenStatements = [];
     this.seenParameters = [];
     this.server = {};
+
+    this._initializationPromise = Promise.resolve(this);
   }
 
   run(statement, parameters) {
@@ -61,7 +64,8 @@ export default class FakeConnection {
   }
 
   initializationCompleted() {
-    return Promise.resolve(this);
+    this.initializationInvoked++;
+    return this._initializationPromise;
   }
 
   isReleasedOnceOnSessionClose() {
@@ -94,4 +98,9 @@ export default class FakeConnection {
     this.server.version = version;
     return this;
   }
+
+  withFailedInitialization(error) {
+    this._initializationPromise = Promise.reject(error);
+    return this;
+  }
 };