Commit and rollback shall cancel unfinished streaming

However session.close will send reset which cancels query execution.

Author: Zhen Li

src/internal/stream-observers.js

@@ -282,9 +282,9 @@ class ResultStreamObserver extends StreamObserver {
   }
 
   /**
-   * Discard pending record stream
+   * Cancel pending record stream
    */
-  discard () {
+  cancel () {
     this._discard = true
   }
 

src/result-rx.js

@@ -134,7 +134,7 @@ export default class RxResult {
       })
 
       if (this._records.observers.length === 0) {
-        result._discard()
+        result._cancel()
       }
 
       result.subscribe({

src/result.js

@@ -85,6 +85,7 @@ class Result {
   summary () {
     return new Promise((resolve, reject) => {
       this._streamObserverPromise.then(o => {
+        o.cancel()
         o.subscribe({
           onCompleted: metadata => resolve(metadata),
           onError: err => reject(err)
@@ -192,8 +193,8 @@ class Result {
    * @protected
    * @since 4.0.0
    */
-  _discard () {
-    this._streamObserverPromise.then(o => o.discard())
+  _cancel () {
+    this._streamObserverPromise.then(o => o.cancel())
   }
 }
 

src/transaction.js

@@ -54,6 +54,7 @@ class Transaction {
     this._onError = this._onErrorCallback.bind(this)
     this._onComplete = this._onCompleteCallback.bind(this)
     this._fetchSize = fetchSize
+    this._results = []
   }
 
   _begin (bookmark, txConfig) {
@@ -86,27 +87,30 @@ class Transaction {
       parameters
     )
 
-    return this._state.run(query, params, {
+    var result = this._state.run(query, params, {
       connectionHolder: this._connectionHolder,
       onError: this._onError,
       onComplete: this._onComplete,
       reactive: this._reactive,
       fetchSize: this._fetchSize
     })
+    this._results.push(result)
+    return result
   }
 
   /**
    * Commits the transaction and returns the result.
    *
    * After committing the transaction can no longer be used.
    *
-   * @returns {Result} New Result
+   * @returns {Promise<void>} An empty promise if committed successfully or error if any error happened during commit.
    */
   commit () {
     const committed = this._state.commit({
       connectionHolder: this._connectionHolder,
       onError: this._onError,
-      onComplete: this._onComplete
+      onComplete: this._onComplete,
+      pendingResults: this._results
     })
     this._state = committed.state
     // clean up
@@ -124,13 +128,15 @@ class Transaction {
    *
    * After rolling back, the transaction can no longer be used.
    *
-   * @returns {Result} New Result
+   * @returns {Promise<void>} An empty promise if rolled back successfully or error if any error happened during
+   * rollback.
    */
   rollback () {
     const rolledback = this._state.rollback({
       connectionHolder: this._connectionHolder,
       onError: this._onError,
-      onComplete: this._onComplete
+      onComplete: this._onComplete,
+      pendingResults: this._results
     })
     this._state = rolledback.state
     // clean up
@@ -170,15 +176,27 @@ class Transaction {
 const _states = {
   // The transaction is running with no explicit success or failure marked
   ACTIVE: {
-    commit: ({ connectionHolder, onError, onComplete }) => {
+    commit: ({ connectionHolder, onError, onComplete, pendingResults }) => {
       return {
-        result: finishTransaction(true, connectionHolder, onError, onComplete),
+        result: finishTransaction(
+          true,
+          connectionHolder,
+          onError,
+          onComplete,
+          pendingResults
+        ),
         state: _states.SUCCEEDED
       }
     },
-    rollback: ({ connectionHolder, onError, onComplete }) => {
+    rollback: ({ connectionHolder, onError, onComplete, pendingResults }) => {
       return {
-        result: finishTransaction(false, connectionHolder, onError, onComplete),
+        result: finishTransaction(
+          false,
+          connectionHolder,
+          onError,
+          onComplete,
+          pendingResults
+        ),
         state: _states.ROLLED_BACK
       }
     },
@@ -188,14 +206,13 @@ const _states = {
       { connectionHolder, onError, onComplete, reactive, fetchSize }
     ) => {
       // RUN in explicit transaction can't contain bookmarks and transaction configuration
+      // No need to include mode and database name as it shall be inclued in begin
       const observerPromise = connectionHolder
         .getConnection()
         .then(conn =>
           conn.protocol().run(statement, parameters, {
             bookmark: Bookmark.empty(),
             txConfig: TxConfig.empty(),
-            mode: connectionHolder.mode(),
-            database: connectionHolder.database(),
             beforeError: onError,
             afterComplete: onComplete,
             reactive: reactive,
@@ -356,22 +373,32 @@ const _states = {
  * @param {ConnectionHolder} connectionHolder
  * @param {function(err:Error): any} onError
  * @param {function(metadata:object): any} onComplete
+ * @param {list<Result>>}pendingResults all run results in this transaction
  */
-function finishTransaction (commit, connectionHolder, onError, onComplete) {
+function finishTransaction (
+  commit,
+  connectionHolder,
+  onError,
+  onComplete,
+  pendingResults
+) {
   const observerPromise = connectionHolder
     .getConnection()
     .then(connection => {
-      if (commit) {
-        return connection.protocol().commitTransaction({
-          beforeError: onError,
-          afterComplete: onComplete
-        })
-      } else {
-        return connection.protocol().rollbackTransaction({
-          beforeError: onError,
-          afterComplete: onComplete
-        })
-      }
+      pendingResults.forEach(r => r.summary())
+      return Promise.all(pendingResults).then(results => {
+        if (commit) {
+          return connection.protocol().commitTransaction({
+            beforeError: onError,
+            afterComplete: onComplete
+          })
+        } else {
+          return connection.protocol().rollbackTransaction({
+            beforeError: onError,
+            afterComplete: onComplete
+          })
+        }
+      })
     })
     .catch(error => new FailedObserver({ error, onError }))
 

test/internal/node/direct.driver.boltkit.test.js

@@ -585,6 +585,99 @@ describe('#stub-direct direct driver with stub server', () => {
     it('v2', () => verifyFailureOnCommit('v2'))
   })
 
+  describe('should cancel stream with result summary method', () => {
+    async function verifyFailureOnCommit (version) {
+      if (!boltStub.supported) {
+        return
+      }
+
+      const server = await boltStub.start(
+        `./test/resources/boltstub/${version}/read_discard.script`,
+        9001
+      )
+
+      const driver = boltStub.newDriver('bolt://127.0.0.1:9001')
+      const session = driver.session({ defaultAccessMode: READ, fetchSize: 2 })
+
+      const result = session.run('MATCH (n) RETURN n.name')
+      await result.summary()
+      const records = (await result).records
+      expect(records.length).toEqual(2)
+      expect(records[0].get(0)).toBe('Bob')
+      expect(records[1].get(0)).toBe('Alice')
+
+      const connectionKey = Object.keys(openConnections(driver))[0]
+      expect(connectionKey).toBeTruthy()
+
+      const connection = openConnections(driver, connectionKey)
+      await session.close()
+
+      // generate a fake fatal error
+      connection._handleFatalError(
+        newError('connection reset', SERVICE_UNAVAILABLE)
+      )
+
+      // expect that the connection to be removed from the pool
+      expect(connectionPool(driver, '127.0.0.1:9001').length).toEqual(0)
+      expect(activeResources(driver, '127.0.0.1:9001')).toBeFalsy()
+      // expect that the connection to be unregistered from the open connections registry
+      expect(openConnections(driver, connectionKey)).toBeFalsy()
+
+      await driver.close()
+      await server.exit()
+    }
+
+    it('v4', () => verifyFailureOnCommit('v4'))
+  })
+
+  describe('should cancel stream with tx commit', () => {
+    async function verifyFailureOnCommit (version) {
+      if (!boltStub.supported) {
+        return
+      }
+
+      const server = await boltStub.start(
+        `./test/resources/boltstub/${version}/read_tx_discard.script`,
+        9001
+      )
+
+      const driver = boltStub.newDriver('bolt://127.0.0.1:9001')
+      const session = driver.session({ defaultAccessMode: READ, fetchSize: 2 })
+      const tx = session.beginTransaction()
+
+      const result = tx.run('MATCH (n) RETURN n.name')
+      await tx.commit()
+
+      // Client will receive a partial result
+      const records = (await result).records
+      expect(records.length).toEqual(2)
+      expect(records[0].get(0)).toBe('Bob')
+      expect(records[1].get(0)).toBe('Alice')
+
+      const connectionKey = Object.keys(openConnections(driver))[0]
+      expect(connectionKey).toBeTruthy()
+
+      const connection = openConnections(driver, connectionKey)
+      await session.close()
+
+      // generate a fake fatal error
+      connection._handleFatalError(
+        newError('connection reset', SERVICE_UNAVAILABLE)
+      )
+
+      // expect that the connection to be removed from the pool
+      expect(connectionPool(driver, '127.0.0.1:9001').length).toEqual(0)
+      expect(activeResources(driver, '127.0.0.1:9001')).toBeFalsy()
+      // expect that the connection to be unregistered from the open connections registry
+      expect(openConnections(driver, connectionKey)).toBeFalsy()
+
+      await driver.close()
+      await server.exit()
+    }
+
+    it('v4', () => verifyFailureOnCommit('v4'))
+  })
+
   function connectionPool (driver, key) {
     return driver._connectionProvider._connectionPool._pools[key]
   }

test/resources/boltstub/v3/acquire_endpoints_self_as_reader.script

@@ -9,7 +9,7 @@ S: SUCCESS {"fields": ["ttl", "servers"]}
    RECORD [9223372036854775807, [{"addresses": ["127.0.0.1:9008"],"role": "WRITE"}, {"addresses": ["127.0.0.1:9001","127.0.0.1:9009","127.0.0.1:9010"], "role": "READ"},{"addresses": ["127.0.0.1:9001","127.0.0.1:9011"], "role": "ROUTE"}]]
    SUCCESS {}
 C: BEGIN {"mode": "r"}
-   RUN "MATCH (n) RETURN n.name AS name" {} {"mode": "r"}
+   RUN "MATCH (n) RETURN n.name AS name" {} {}
    PULL_ALL
 S: SUCCESS {}
    SUCCESS {"fields": ["name"]}

test/resources/boltstub/v3/read_tx.script

@@ -5,7 +5,7 @@
 
 C: BEGIN {"mode": "r"}
 S: SUCCESS {}
-C: RUN "MATCH (n) RETURN n.name" {} {"mode": "r"}
+C: RUN "MATCH (n) RETURN n.name" {} {}
    PULL_ALL
 S: SUCCESS {"fields": ["n.name"]}
    RECORD ["Bob"]

test/resources/boltstub/v3/read_tx_dead.script

@@ -3,7 +3,7 @@
 !: AUTO RESET
 
 C: BEGIN {"mode": "r"}
-   RUN "MATCH (n) RETURN n.name" {} {"mode": "r"}
+   RUN "MATCH (n) RETURN n.name" {} {}
    PULL_ALL
 S: SUCCESS {"fields": []}
    <EXIT>

test/resources/boltstub/v3/read_tx_with_bookmarks.script

@@ -4,7 +4,7 @@
 !: AUTO GOODBYE
 
 C: BEGIN {"bookmarks": ["neo4j:bookmark:v1:tx42"], "mode": "r"}
-C: RUN "MATCH (n) RETURN n.name AS name" {} {"mode": "r"}
+C: RUN "MATCH (n) RETURN n.name AS name" {} {}
    PULL_ALL
 S: SUCCESS {}
    SUCCESS {"fields": ["name"]}

test/resources/boltstub/v4/read_discard.script

@@ -0,0 +1,13 @@
+!: BOLT 4
+!: AUTO HELLO
+!: AUTO RESET
+!: AUTO GOODBYE
+
+C: RUN "MATCH (n) RETURN n.name" {} {"mode": "r"}
+   PULL {"n": 2}
+S: SUCCESS {"fields": ["n.name"]}
+   RECORD ["Bob"]
+   RECORD ["Alice"]
+   SUCCESS {"has_more":true}
+C: DISCARD {"n": -1}
+S: SUCCESS {}