Prevent resolve or iterate over already consumed Result

This `result.then` and `for await (const r of result)` in already consumed results could block for ever since no new event will came from wire.

Blocking this kind of access improves not correct access to the Result object.

This change also includes the addition of `Result.isOpen()`

Author: bigmontz

packages/core/src/result.ts

@@ -21,6 +21,7 @@ import ResultSummary from './result-summary'
 import Record from './record'
 import { Query, PeekableAsyncIterator } from './types'
 import { observer, util, connectionHolder } from './internal'
+import { newError } from './error'
 
 const { EMPTY_CONNECTION_HOLDER } = connectionHolder
 
@@ -208,6 +209,9 @@ class Result implements Promise<QueryResult> {
    * @return {Promise} new Promise.
    */
   private _getOrCreatePromise(): Promise<QueryResult> {
+    if (!this.isOpen()) {
+      return Promise.reject(newError('Result is already consumed'))
+    }
     if (!this._p) {
       this._p = new Promise((resolve, reject) => {
         const records: Record[] = []
@@ -238,6 +242,13 @@ class Result implements Promise<QueryResult> {
    * @returns {PeekableAsyncIterator<Record, ResultSummary>} The async iterator for the Results
    */
   [Symbol.asyncIterator](): PeekableAsyncIterator<Record, ResultSummary> {
+    if (!this.isOpen()) {
+      const error = newError('Result is already consumed')
+      return {
+        next: () => Promise.reject(error),
+        peek: () => Promise.reject(error),
+      }
+    }
     const state: {
       paused: boolean,
       firstRun: boolean,
@@ -362,6 +373,10 @@ class Result implements Promise<QueryResult> {
       .catch(() => {})
   }
 
+  isOpen (): boolean {
+    return this._summary === null && this._error === null
+  }
+
   /**
    * Stream records to observer as they come in, this is a more efficient method
    * of handling the results, and allows you to handle arbitrarily large results.

packages/core/src/transaction.ts

@@ -600,7 +600,7 @@ function finishTransaction(
     .then(connection => {
       onConnection()
       pendingResults.forEach(r => r._cancel())
-      return Promise.all(pendingResults).then(results => {
+      return Promise.all(pendingResults.map(result => result.summary())).then(results => {
         if (connection) {
           if (commit) {
             return connection.protocol().commitTransaction({

packages/core/test/result.test.ts

@@ -451,6 +451,22 @@ describe('Result', () => {
         result.catch(() => { }).finally(done)
       })
 
+      it.each([
+        ['success', async (stream: any) => stream.onCompleted({})],
+        //['error', async (stream: any) => stream.onError(new Error('error'))],
+      ])('should thrown over an consumed result [%s]', async(_, completeStream) => {
+        completeStream(streamObserverMock)
+
+        await result.summary().catch(() => {})
+
+        try {
+          await result
+          expect('not finish iteration over consumed result').toBe(true)
+        } catch (e) {
+          expect(e).toEqual(newError('Result is already consumed'))
+        }
+      })
+
       describe.each([
         ['query', {}, { query: 'query', parameters: {} }],
         ['query', { a: 1 }, { query: 'query', parameters: { a: 1 } }],
@@ -860,6 +876,26 @@ describe('Result', () => {
         ])
       })
 
+      it.each([
+        ['success', async (stream: any) => stream.onCompleted({})],
+        ['error', async (stream: any) => stream.onError(new Error('error'))],
+      ])('should thrown on iterate over an consumed result [%s]', async(_, completeStream) => {
+        completeStream(streamObserverMock)
+
+        await result.summary().catch(() => {})
+
+        try {
+          for await (const _ of result) {
+            expect('not iterate over consumed result').toBe(true)
+          }
+          expect('not finish iteration over consumed result').toBe(true)
+        } catch (e) {
+          expect(e).toEqual(newError('Result is already consumed'))
+        }
+
+        expect('not finish iteration over consumed result')
+      })
+
       describe('.return()', () => {
         it('should finished the operator when it get called', async () => {
           const keys = ['a', 'b']
@@ -1205,6 +1241,30 @@ describe('Result', () => {
         })
       })
     })
+
+    describe('.isOpen()', () => {
+      it('should return true when the stream is open', async () => {
+        await result._subscribe({}).catch(() => {})
+        
+        expect(result.isOpen()).toBe(true)
+      })
+
+      it('should return false when the stream is closed', async () => {
+        streamObserverMock.onCompleted({})
+
+        await result._subscribe({}).catch(() => {})
+
+        expect(result.isOpen()).toBe(false)
+      })
+
+      it('should return false when the stream is failed', async () => {
+        streamObserverMock.onError(new Error('test'))
+
+        await result._subscribe({}).catch(() => {})
+
+        expect(result.isOpen()).toBe(false)
+      })
+    })
   })
 
   describe.each([
@@ -1319,6 +1379,19 @@ describe('Result', () => {
         })
       })
     })
+
+    describe('isOpen()', () => {
+      it('should be true', () => {
+        expect(result.isOpen()).toBe(true)
+      })
+
+      it('should be false after any interactio with the stream', async () => {
+        const it = result[Symbol.asyncIterator]()
+        await it.next()
+
+        expect(result.isOpen()).toBe(false)
+      })
+    })
   })
 
   describe.each([
@@ -1381,6 +1454,30 @@ describe('Result', () => {
       })
     })
 
+    describe('.isOpen()', () => {
+      it('should be true', async () => {
+        expect(result.isOpen()).toBe(true)
+
+        // consume the stream to avoid unhaddled errors
+        try {
+          await result.summary()
+        } catch (error) {
+        }
+      })
+
+      it('should be false after any interactio with the stream', async () => {
+        const it = result[Symbol.asyncIterator]()
+        
+        try {
+          await it.next()
+        } catch (error) {
+          // this's fine
+        }
+
+        expect(result.isOpen()).toBe(false)
+      })
+    })
+
     function shouldReturnRejectedPromiseWithTheExpectedError<T>(
       supplier: () => Promise<T>
     ) {

packages/testkit-backend/src/controller/local.js

@@ -63,7 +63,7 @@ export default class LocalController extends Controller {
         const id = this._contexts.get(contextId).addError(e)
         this._writeResponse(contextId, newResponse('DriverError', {
           id,
-          msg: e.message + ' (' + e.code + ')',
+          msg: e.message,
           code: e.code
         }))
       }

packages/testkit-backend/src/skipped-tests/common.js

@@ -138,10 +138,10 @@ const skippedTests = [
       'stub.iteration.test_result_list.TestResultList.test_session_run_result_list_pulls_all_records_at_once_next_before_list'
     )
   ),
-  skip(
-    'Results are always valid but don\'t return records when out of scope',
-    ifStartsWith('stub.iteration.test_result_scope.TestResultScope.')
-  ),
+  // skip(
+  //   'Results are always valid but don\'t return records when out of scope',
+  //   ifStartsWith('stub.iteration.test_result_scope.TestResultScope.')
+  // ),
   skip(
     'Driver (still) allows explicit managing of managed transaction',
     ifEquals('stub.tx_lifetime.test_tx_lifetime.TestTxLifetime.test_managed_tx_raises_tx_managed_exec')