Make the stress test more reliable

 Avoid session.run usage
 Treat the session.close() exception in the command.
 Check the number of commands run

Author: bigmontz

test/stress.test.js

@@ -97,27 +97,32 @@ describe('#integration stress tests', () => {
     await driver.close()
   })
 
-  it('basic', done => {
-    const context = new Context(driver, LOGGING_ENABLED)
-    const commands = createCommands(context)
+  it(
+    'basic',
+    done => {
+      const context = new Context(driver, LOGGING_ENABLED)
+      const commands = createCommands(context)
 
-    console.time('Basic-stress-test')
-    parallelLimit(commands, TEST_MODE.parallelism, error => {
-      console.timeEnd('Basic-stress-test')
+      console.time('Basic-stress-test')
+      parallelLimit(commands, TEST_MODE.parallelism, error => {
+        console.timeEnd('Basic-stress-test')
 
-      console.log('Read statistics: ', context.readServersWithQueryCount)
-      console.log('Write statistics: ', context.writeServersWithQueryCount)
+        console.log('Read statistics: ', context.readServersWithQueryCount)
+        console.log('Write statistics: ', context.writeServersWithQueryCount)
 
-      if (error) {
-        done.fail(error)
-      }
+        if (error) {
+          done.fail(error)
+        }
 
-      verifyServers(context)
-        .then(() => verifyNodeCount(context))
-        .then(() => done())
-        .catch(error => done.fail(error))
-    })
-  })
+        verifyServers(context)
+          .then(() => verifyCommandsRun(context, TEST_MODE.commandsCount))
+          .then(() => verifyNodeCount(context))
+          .then(() => done())
+          .catch(error => done.fail(error))
+      })
+    },
+    TEST_MODE.maxRunTimeMs
+  )
 
   function createCommands (context) {
     const uniqueCommands = createUniqueCommands(context)
@@ -276,7 +281,7 @@ describe('#integration stress tests', () => {
           context.log(commandId, 'Query completed successfully')
 
           return session.close().then(() => {
-            const possibleError = verifyQueryResult(result)
+            const possibleError = verifyQueryResult(result, context)
             callback(possibleError)
           })
         })
@@ -316,10 +321,19 @@ describe('#integration stress tests', () => {
           context.queryCompleted(result, accessMode, session.lastBookmark())
           context.log(commandId, 'Transaction function executed successfully')
 
-          return session.close().then(() => {
-            const possibleError = verifyQueryResult(result)
-            callback(possibleError)
-          })
+          return session
+            .close()
+            .then(() => {
+              const possibleError = verifyQueryResult(result, context)
+              callback(possibleError)
+            })
+            .catch(error => {
+              context.log(
+                commandId,
+                `Error closing the session ${JSON.stringify(error)}`
+              )
+              callback(error)
+            })
         })
         .catch(error => {
           context.log(
@@ -355,7 +369,7 @@ describe('#integration stress tests', () => {
 
       tx.run(query, params)
         .then(result => {
-          let commandError = verifyQueryResult(result)
+          let commandError = verifyQueryResult(result, context)
 
           tx.commit()
             .catch(commitError => {
@@ -388,10 +402,13 @@ describe('#integration stress tests', () => {
     }
   }
 
-  function verifyQueryResult (result) {
+  function verifyQueryResult (result, context) {
     if (!result) {
       return new Error('Received undefined result')
-    } else if (result.records.length === 0) {
+    } else if (
+      result.records.length === 0 &&
+      context.writeCommandsRun < TEST_MODE.parallelism
+    ) {
       // it is ok to receive no nodes back for read queries at the beginning of the test
       return null
     } else if (result.records.length === 1) {
@@ -424,6 +441,14 @@ describe('#integration stress tests', () => {
     return null
   }
 
+  function verifyCommandsRun (context, expectedCommandsRun) {
+    if (context.commandsRun !== expectedCommandsRun) {
+      throw new Error(
+        `Unexpected commands run: ${context.commandsRun}, expected: ${expectedCommandsRun}`
+      )
+    }
+  }
+
   function verifyNodeCount (context) {
     const expectedNodeCount = context.createdNodesCount
 
@@ -507,10 +532,10 @@ describe('#integration stress tests', () => {
 
   function fetchClusterAddresses (context) {
     const session = context.driver.session()
-    return session.run('CALL dbms.cluster.overview()').then(result =>
-      session.close().then(() => {
+    return session
+      .readTransaction(tx => tx.run('CALL dbms.cluster.overview()'))
+      .then(result => {
         const records = result.records
-
         const supportsMultiDb = protocolVersion >= 4.0
         const followers = supportsMultiDb
           ? addressesForMultiDb(records, 'FOLLOWER')
@@ -519,9 +544,10 @@ describe('#integration stress tests', () => {
           ? addressesForMultiDb(records, 'READ_REPLICA')
           : addressesWithRole(records, 'READ_REPLICA')
 
-        return new ClusterAddresses(followers, readReplicas)
+        return session
+          .close()
+          .then(() => new ClusterAddresses(followers, readReplicas))
       })
-    )
   }
 
   function addressesForMultiDb (records, role, db = 'neo4j') {
@@ -631,6 +657,20 @@ describe('#integration stress tests', () => {
       this.protocolVersion = null
     }
 
+    get commandsRun () {
+      return [
+        ...Object.values(this.readServersWithQueryCount),
+        ...Object.values(this.writeServersWithQueryCount)
+      ].reduce((a, b) => a + b, 0)
+    }
+
+    get writeCommandsRun () {
+      return [...Object.values(this.writeServersWithQueryCount)].reduce(
+        (a, b) => a + b,
+        0
+      )
+    }
+
     queryCompleted (result, accessMode, bookmark) {
       const serverInfo = result.summary.server
       this.protocolVersion = serverInfo.protocolVersion