Avoid to send reset when it is not needed

The `RESET` should be send when a failure occurs or whenever the connection is being sent back
to the pool with a pending request running, i.e. the bolt server is not in the `READY` state.

These changes also affect the `verifyConnectivity` and `getServerInfo` implementation.
The `RESET` message is not sent in these methods if it is a newly created connection.

Author: bigmontz

packages/bolt-connection/src/connection-provider/connection-provider-pooled.js

@@ -71,7 +71,11 @@ export default class PooledConnectionProvider extends ConnectionProvider {
    */
   _createConnection (address, release) {
     return this._createChannelConnection(address).then(connection => {
-      connection._release = () => release(address, connection)
+      connection._firstUsage = true
+      connection._release = () => {
+        connection._firstUsage = false
+        return release(address, connection)
+      }
       this._openConnections[connection.id] = connection
       return connection
         .connect(this._userAgent, this._authToken)
@@ -119,7 +123,9 @@ export default class PooledConnectionProvider extends ConnectionProvider {
     const connection = await this._connectionPool.acquire(address)
     const serverInfo = new ServerInfo(connection.server, connection.protocol().version)
     try {
-      await connection.resetAndFlush()
+      if (!connection._firstUsage) {
+        await connection.resetAndFlush()
+      }
     } finally {
       await connection._release()
     }

packages/bolt-connection/src/connection/connection-channel.js

@@ -122,6 +122,7 @@ export default class ChannelConnection extends Connection {
   ) {
     super(errorHandler)
 
+    this._reseting = false
     this._id = idGenerator++
     this._address = address
     this._server = { address: address.asHostPort() }
@@ -303,7 +304,7 @@ export default class ChannelConnection extends Connection {
    */
   resetAndFlush () {
     return new Promise((resolve, reject) => {
-      this._protocol.reset({
+      this._reset({
         onError: error => {
           if (this._isBroken) {
             // handling a fatal error, no need to raise a protocol violation
@@ -327,7 +328,7 @@ export default class ChannelConnection extends Connection {
       return
     }
 
-    this._protocol.reset({
+    this._reset({
       onError: () => {
         this._protocol.resetFailure()
       },
@@ -337,6 +338,23 @@ export default class ChannelConnection extends Connection {
     })
   }
 
+  _reset(observer) {
+    if (this._reseting) {
+      observer.onComplete()
+      return
+    }
+    this._reseting = true
+    this._protocol.reset({
+      onError: error => {
+        this._reseting = false
+        observer.onError(error)
+      }, onComplete: () => {
+        this._reseting = false
+        observer.onComplete()
+      }
+    })
+  }
+
   /*
    * Pop next pending observer form the list of observers and make it current observer.
    * @protected

packages/bolt-connection/src/connection/connection-delegate.js

@@ -83,6 +83,10 @@ export default class DelegateConnection extends Connection {
     return this._delegate.resetAndFlush()
   }
 
+  hasOngoingObservableRequests () {
+    return this._delegate.hasOngoingObservableRequests()
+  }
+
   close () {
     return this._delegate.close()
   }

packages/bolt-connection/src/connection/connection.js

@@ -103,6 +103,10 @@ export default class Connection {
     throw new Error('not implemented')
   }
 
+  hasOngoingObservableRequests () {
+    throw new Error('not implemented')
+  }
+
   /**
    * Call close on the channel.
    * @returns {Promise<void>} - A promise that will be resolved when the connection is closed.

packages/bolt-connection/test/connection-provider/connection-provider-routing.test.js

@@ -2590,7 +2590,26 @@ describe('#unit RoutingConnectionProvider', () => {
         expect(connections[0]._release).toHaveBeenCalled()
         expect(connections[0]._release.mock.invocationCallOrder[0])
           .toBeGreaterThan(connections[0].resetAndFlush.mock.invocationCallOrder[0])
-      
+      })
+
+      it('should not call resetAndFlush for newly created connections', async () => {
+        const { connectionProvider, routingTable, seenConnectionsPerAddress, pool } = setup({ newConnection: true })
+        const acquireSpy = jest.spyOn(pool, 'acquire')
+
+        await connectionProvider.verifyConnectivityAndGetServerInfo({ database, accessMode })
+
+        const targetServers = accessMode === WRITE ? routingTable.writers : routingTable.readers
+        const address = targetServers[0]
+        expect(acquireSpy).toHaveBeenCalledWith(address)
+
+        const connections = seenConnectionsPerAddress.get(address)
+
+        // verifying resetAndFlush was not called
+        expect(connections[0].resetAndFlush).not.toHaveBeenCalled()
+
+        // extra checks
+        expect(connections.length).toBe(1)
+        expect(connections[0]._release).toHaveBeenCalled()
       })
 
       it('should not acquire, resetAndFlush and release connections for sever with the other access mode', async () => {
@@ -2766,7 +2785,7 @@ describe('#unit RoutingConnectionProvider', () => {
         })
       })
 
-      function setup({ resetAndFlush, releaseMock } = {}) {
+      function setup({ resetAndFlush, releaseMock, newConnection } = { }) {
         const routingTable = newRoutingTable(
           database || null,
           [server1, server2],
@@ -2784,6 +2803,7 @@ describe('#unit RoutingConnectionProvider', () => {
               seenConnectionsPerAddress.set(address, [])
             }
             const connection = new FakeConnection(address, release, 'version', protocolVersion, server)
+            connection._firstUsage = !!newConnection
             if (resetAndFlush) {
               connection.resetAndFlush = resetAndFlush
             }

packages/bolt-connection/test/connection/connection-channel.test.js

@@ -19,6 +19,7 @@
 
 import ChannelConnection from '../../src/connection/connection-channel'
 import { int, internal, newError } from 'neo4j-driver-core'
+import { observer } from 'neo4j-driver-core/types/internal'
 
 const {
   serverAddress: { ServerAddress },
@@ -260,6 +261,75 @@ describe('ChannelConnection', () => {
         expect(protocol.reset).toHaveBeenCalled()
         expect(protocol.resetFailure).toHaveBeenCalled()
       })
+
+      it('should not call protocol.reset() when there is an ongoing reset', () => {
+        const channel = {
+          _open: true
+        }
+
+        const protocol = {
+          reset: jest.fn(),
+          resetFailure: jest.fn()
+        }
+        const protocolSupplier = () => protocol
+        const connection = spyOnConnectionChannel({ channel, protocolSupplier })
+
+        connection._resetOnFailure()
+
+        expect(protocol.reset).toHaveBeenCalledTimes(1)
+        expect(protocol.resetFailure).not.toHaveBeenCalled()
+
+        connection._resetOnFailure()
+
+        expect(protocol.reset).toHaveBeenCalledTimes(1)
+        expect(protocol.resetFailure).toHaveBeenCalledTimes(1)
+      })
+
+      it('should call protocol.reset() when after a previous reset completed', () => {
+        const channel = {
+          _open: true
+        }
+
+        const protocol = {
+          reset: jest.fn(observer => observer.onComplete()),
+          resetFailure: jest.fn()
+        }
+        const protocolSupplier = () => protocol
+        const connection = spyOnConnectionChannel({ channel, protocolSupplier })
+
+        connection._resetOnFailure()
+
+        expect(protocol.reset).toHaveBeenCalledTimes(1)
+        expect(protocol.resetFailure).toHaveBeenCalledTimes(1)
+
+        connection._resetOnFailure()
+
+        expect(protocol.reset).toHaveBeenCalledTimes(2)
+        expect(protocol.resetFailure).toHaveBeenCalledTimes(2)
+      })
+
+      it('should call protocol.reset() when after a previous reset fail', () => {
+        const channel = {
+          _open: true
+        }
+
+        const protocol = {
+          reset: jest.fn(observer => observer.onError(new Error('some error'))),
+          resetFailure: jest.fn()
+        }
+        const protocolSupplier = () => protocol
+        const connection = spyOnConnectionChannel({ channel, protocolSupplier })
+
+        connection._resetOnFailure()
+
+        expect(protocol.reset).toHaveBeenCalledTimes(1)
+        expect(protocol.resetFailure).toHaveBeenCalledTimes(1)
+
+        connection._resetOnFailure()
+
+        expect(protocol.reset).toHaveBeenCalledTimes(2)
+        expect(protocol.resetFailure).toHaveBeenCalledTimes(2)
+      })
     })
 
     describe('when connection is not open', () => {
@@ -286,6 +356,106 @@ describe('ChannelConnection', () => {
     })
   })
 
+  describe('.resetAndFlush()', () => {
+    it('should call protocol.reset() onComplete', async () => {
+      const channel = {
+        _open: true
+      }
+
+      const protocol = {
+        reset: jest.fn(observer => observer.onComplete()),
+        resetFailure: jest.fn()
+      }
+      const protocolSupplier = () => protocol
+      const connection = spyOnConnectionChannel({ channel, protocolSupplier })
+
+      await connection.resetAndFlush().catch(() => {})
+
+      expect(protocol.reset).toHaveBeenCalled()
+    })
+
+    it('should call protocol.reset() onError', async () => {
+      const channel = {
+        _open: true
+      }
+
+      const protocol = {
+        reset: jest.fn(observer => observer.onError()),
+        resetFailure: jest.fn()
+      }
+      const protocolSupplier = () => protocol
+      const connection = spyOnConnectionChannel({ channel, protocolSupplier })
+
+      await connection.resetAndFlush().catch(() => {})
+
+      expect(protocol.reset).toHaveBeenCalled()
+    })
+
+    it('should not call protocol.reset() when there is an ongoing reset', async () => {
+      const channel = {
+        _open: true
+      }
+
+      const protocol = {
+        reset: jest.fn(),
+        resetFailure: jest.fn()
+      }
+      const protocolSupplier = () => protocol
+      const connection = spyOnConnectionChannel({ channel, protocolSupplier })
+
+      // to not block since the reset will never complete
+      connection.resetAndFlush()
+
+      expect(protocol.reset).toHaveBeenCalledTimes(1)
+
+      await connection.resetAndFlush()
+
+      expect(protocol.reset).toHaveBeenCalledTimes(1)
+    })
+
+    it('should call protocol.reset() when after a previous reset completed', async () => {
+      const channel = {
+        _open: true
+      }
+
+      const protocol = {
+        reset: jest.fn(observer => observer.onComplete()),
+        resetFailure: jest.fn()
+      }
+      const protocolSupplier = () => protocol
+      const connection = spyOnConnectionChannel({ channel, protocolSupplier })
+
+      await connection.resetAndFlush()
+
+      expect(protocol.reset).toHaveBeenCalledTimes(1)
+
+      await connection.resetAndFlush()
+
+      expect(protocol.reset).toHaveBeenCalledTimes(2)
+    })
+
+    it('should call protocol.reset() when after a previous reset fail', async () => {
+      const channel = {
+        _open: true
+      }
+
+      const protocol = {
+        reset: jest.fn(observer => observer.onError(new Error('some error'))),
+        resetFailure: jest.fn()
+      }
+      const protocolSupplier = () => protocol
+      const connection = spyOnConnectionChannel({ channel, protocolSupplier })
+
+      await connection.resetAndFlush().catch(() => {})
+
+      expect(protocol.reset).toHaveBeenCalledTimes(1)
+
+      await connection.resetAndFlush().catch(() => {})
+
+      expect(protocol.reset).toHaveBeenCalledTimes(2)
+    })
+  })
+
   function spyOnConnectionChannel ({
     channel,
     errorHandler,

packages/core/src/connection.ts

@@ -93,6 +93,14 @@ class Connection {
     throw Error('Not implemented')
   }
 
+  /**
+   * Checks if there is an ongoing request being handled
+   * @return {boolean} `true` if there is an ongoing request being handled
+   */
+  hasOngoingObservableRequests (): boolean {
+    throw Error('Not implemented')
+  }
+
   /**
    * Call close on the channel.
    * @returns {Promise<void>} - A promise that will be resolved when the connection is closed.

packages/core/src/internal/connection-holder.ts

@@ -196,7 +196,7 @@ class ConnectionHolder implements ConnectionHolderInterface {
     this._connectionPromise = this._connectionPromise
       .then((connection?: Connection|void) => {
         if (connection) {
-          if (connection.isOpen()) {
+          if (connection.isOpen() && connection.hasOngoingObservableRequests()) {
             return connection
               .resetAndFlush()
               .catch(ignoreError)

packages/core/test/session.test.ts

@@ -186,6 +186,27 @@ describe('session', () => {
     })
   }, 70000)
 
+  it('close should reset connection if there is an ongoing request ', async () => {
+    const connection = newFakeConnection()
+    const resetAndFlushSpy = jest.spyOn(connection, 'resetAndFlush')
+    const session = newSessionWithConnection(connection)
+
+    await session.close()
+
+    expect(resetAndFlushSpy).toHaveBeenCalledTimes(1)
+  }, 70000)
+
+  it('close should not reset connection if there is not an ongoing request ', async () => {
+    const connection = newFakeConnection()
+    connection.hasOngoingObservableRequests = () => false
+    const resetAndFlushSpy = jest.spyOn(connection, 'resetAndFlush')
+    const session = newSessionWithConnection(connection)
+
+    await session.close()
+
+    expect(resetAndFlushSpy).not.toHaveBeenCalled()
+  }, 70000)
+
   it('should close transaction executor', done => {
     const session = newSessionWithConnection(newFakeConnection())
 

packages/core/test/utils/connection.fake.ts

@@ -181,6 +181,10 @@ export default class FakeConnection extends Connection {
     this._open = false
     return this
   }
+
+  hasOngoingObservableRequests(): boolean {
+    return true
+  }
 }
 
 function mockResultStreamObserverWithError (query: string, parameters: any | undefined, error: Error) {