TransactionExecutor checks if tx is active

To make sure it does not try to commit transaction that has already been
committed or rolled back.

This commit also adds a new API function `Transaction#isOpen()` which checks
internal transaction state and returns `true` when transaction has not
been committed/rolled back and `false` otherwise.

Author: lutovich

src/v1/internal/transaction-executor.js

@@ -86,14 +86,21 @@ export default class TransactionExecutor {
       const resultPromise = Promise.resolve(transactionWorkResult);
 
       resultPromise.then(result => {
-        // transaction work returned resolved promise, try to commit the transaction
-        tx.commit().then(() => {
-          // transaction was committed, return result to the user
+        if (tx.isOpen()) {
+          // transaction work returned resolved promise and transaction has not been committed/rolled back
+          // try to commit the transaction
+          tx.commit().then(() => {
+            // transaction was committed, return result to the user
+            resolve(result);
+          }).catch(error => {
+            // transaction failed to commit, propagate the failure
+            reject(error);
+          });
+        } else {
+          // transaction work returned resolved promise and transaction is already committed/rolled back
+          // return the result returned by given transaction work
           resolve(result);
-        }).catch(error => {
-          // transaction failed to commit, propagate the failure
-          reject(error);
-        });
+        }
       }).catch(error => {
         // transaction work returned rejected promise, propagate the failure
         reject(error);

src/v1/transaction.js

@@ -61,7 +61,7 @@ class Transaction {
    * or with the statement and parameters as separate arguments.
    * @param {mixed} statement - Cypher statement to execute
    * @param {Object} parameters - Map with parameters to use in statement
-   * @return {Result} - New Result
+   * @return {Result} New Result
    */
   run(statement, parameters) {
     if(typeof statement === 'object' && statement.text) {
@@ -78,7 +78,7 @@ class Transaction {
    *
    * After committing the transaction can no longer be used.
    *
-   * @returns {Result} - New Result
+   * @returns {Result} New Result
    */
   commit() {
     let committed = this._state.commit(this._connectionHolder, new _TransactionStreamObserver(this));
@@ -93,7 +93,7 @@ class Transaction {
    *
    * After rolling back, the transaction can no longer be used.
    *
-   * @returns {Result} - New Result
+   * @returns {Result} New Result
    */
   rollback() {
     let committed = this._state.rollback(this._connectionHolder, new _TransactionStreamObserver(this));
@@ -103,8 +103,16 @@ class Transaction {
     return committed.result;
   }
 
+  /**
+   * Check if this transaction is active, which means commit and rollback did not happen.
+   * @return {boolean} <code>true</code> when not committed and not rolled back, <code>false</code> otherwise.
+   */
+  isOpen() {
+    return this._state == _states.ACTIVE;
+  }
+
   _onError() {
-    if (this._state == _states.ACTIVE) {
+    if (this.isOpen()) {
       // attempt to rollback, useful when Transaction#run() failed
       return this.rollback().catch(ignoredError => {
         // ignore all errors because it is best effort and transaction might already be rolled back

test/internal/transaction-executor.test.js

@@ -317,6 +317,10 @@ class FakeTransaction {
     this._commitErrorCode = commitErrorCode;
   }
 
+  isOpen() {
+    return true;
+  }
+
   commit() {
     if (this._commitErrorCode) {
       return Promise.reject(error(this._commitErrorCode));

test/v1/session.test.js

@@ -28,13 +28,13 @@ describe('session', () => {
 
   let driver;
   let session;
-  let server;
+  let serverMetadata;
   let originalTimeout;
 
   beforeEach(done => {
     driver = neo4j.driver('bolt://localhost', neo4j.auth.basic('neo4j', 'neo4j'));
     driver.onCompleted = meta => {
-      server = meta['server'];
+      serverMetadata = meta['server'];
     };
     session = driver.session();
     originalTimeout = jasmine.DEFAULT_TIMEOUT_INTERVAL;
@@ -228,11 +228,7 @@ describe('session', () => {
   });
 
   it('should expose server info on successful query', done => {
-    //lazy way of checking the version number
-    //if server has been set we know it is at least
-    //3.1 (todo actually parse the version string)
-    if (!server) {
-      done();
+    if (!serverIs31OrLater(done)) {
       return;
     }
 
@@ -251,14 +247,10 @@ describe('session', () => {
   });
 
   it('should expose execution time information when using 3.1 and onwards', done => {
-
-    //lazy way of checking the version number
-    //if server has been set we know it is at least
-    //3.1 (todo actually parse the version string)
-    if (!server) {
-      done();
+    if (!serverIs31OrLater(done)) {
       return;
     }
+
     // Given
     const statement = 'UNWIND range(1,10000) AS n RETURN n AS number';
     // When & Then
@@ -632,6 +624,184 @@ describe('session', () => {
     });
   });
 
+  it('should commit read transaction', done => {
+    if (!serverIs31OrLater(done)) {
+      return;
+    }
+
+    const bookmarkBefore = session.lastBookmark();
+    const resultPromise = session.readTransaction(tx => tx.run('RETURN 42 AS answer'));
+
+    resultPromise.then(result => {
+      expect(result.records.length).toEqual(1);
+      expect(result.records[0].get('answer').toNumber()).toEqual(42);
+
+      const bookmarkAfter = session.lastBookmark();
+      verifyBookmark(bookmarkAfter);
+      expect(bookmarkAfter).not.toEqual(bookmarkBefore);
+
+      done();
+    });
+  });
+
+  it('should commit write transaction', done => {
+    if (!serverIs31OrLater(done)) {
+      return;
+    }
+
+    const bookmarkBefore = session.lastBookmark();
+    const resultPromise = session.writeTransaction(tx => tx.run('CREATE (n:Node {id: 42}) RETURN n.id AS answer'));
+
+    resultPromise.then(result => {
+      expect(result.records.length).toEqual(1);
+      expect(result.records[0].get('answer').toNumber()).toEqual(42);
+      expect(result.summary.counters.nodesCreated()).toEqual(1);
+
+      const bookmarkAfter = session.lastBookmark();
+      verifyBookmark(bookmarkAfter);
+      expect(bookmarkAfter).not.toEqual(bookmarkBefore);
+
+      countNodes('Node', 'id', 42).then(count => {
+        expect(count).toEqual(1);
+        done();
+      });
+    });
+  });
+
+  it('should not commit already committed read transaction', done => {
+    if (!serverIs31OrLater(done)) {
+      return;
+    }
+
+    const resultPromise = session.readTransaction(tx => {
+      return new Promise((resolve, reject) => {
+        tx.run('RETURN 42 AS answer').then(result => {
+          tx.commit().then(() => {
+            resolve({result: result, bookmark: session.lastBookmark()});
+          }).catch(error => reject(error));
+        }).catch(error => reject(error));
+      });
+    });
+
+    resultPromise.then(outcome => {
+      const bookmark = outcome.bookmark;
+      const result = outcome.result;
+
+      verifyBookmark(bookmark);
+      expect(session.lastBookmark()).toEqual(bookmark); // expect bookmark to not change
+
+      expect(result.records.length).toEqual(1);
+      expect(result.records[0].get('answer').toNumber()).toEqual(42);
+
+      done();
+    });
+  });
+
+  it('should not commit already committed write transaction', done => {
+    if (!serverIs31OrLater(done)) {
+      return;
+    }
+
+    const resultPromise = session.readTransaction(tx => {
+      return new Promise((resolve, reject) => {
+        tx.run('CREATE (n:Node {id: 42}) RETURN n.id AS answer').then(result => {
+          tx.commit().then(() => {
+            resolve({result: result, bookmark: session.lastBookmark()});
+          }).catch(error => reject(error));
+        }).catch(error => reject(error));
+      });
+    });
+
+    resultPromise.then(outcome => {
+      const bookmark = outcome.bookmark;
+      const result = outcome.result;
+
+      verifyBookmark(bookmark);
+      expect(session.lastBookmark()).toEqual(bookmark); // expect bookmark to not change
+
+      expect(result.records.length).toEqual(1);
+      expect(result.records[0].get('answer').toNumber()).toEqual(42);
+      expect(result.summary.counters.nodesCreated()).toEqual(1);
+
+      countNodes('Node', 'id', 42).then(count => {
+        expect(count).toEqual(1);
+        done();
+      });
+    });
+  });
+
+  it('should not commit rolled back read transaction', done => {
+    if (!serverIs31OrLater(done)) {
+      return;
+    }
+
+    const bookmarkBefore = session.lastBookmark();
+    const resultPromise = session.readTransaction(tx => {
+      return new Promise((resolve, reject) => {
+        tx.run('RETURN 42 AS answer').then(result => {
+          tx.rollback().then(() => {
+            resolve(result);
+          }).catch(error => reject(error));
+        }).catch(error => reject(error));
+      });
+    });
+
+    resultPromise.then(result => {
+      expect(result.records.length).toEqual(1);
+      expect(result.records[0].get('answer').toNumber()).toEqual(42);
+      expect(session.lastBookmark()).toBe(bookmarkBefore); // expect bookmark to not change
+
+      done();
+    });
+  });
+
+  it('should not commit rolled back write transaction', done => {
+    if (!serverIs31OrLater(done)) {
+      return;
+    }
+
+    const bookmarkBefore = session.lastBookmark();
+    const resultPromise = session.readTransaction(tx => {
+      return new Promise((resolve, reject) => {
+        tx.run('CREATE (n:Node {id: 42}) RETURN n.id AS answer').then(result => {
+          tx.rollback().then(() => {
+            resolve(result);
+          }).catch(error => reject(error));
+        }).catch(error => reject(error));
+      });
+    });
+
+    resultPromise.then(result => {
+      expect(result.records.length).toEqual(1);
+      expect(result.records[0].get('answer').toNumber()).toEqual(42);
+      expect(result.summary.counters.nodesCreated()).toEqual(1);
+      expect(session.lastBookmark()).toBe(bookmarkBefore); // expect bookmark to not change
+
+      countNodes('Node', 'id', 42).then(count => {
+        expect(count).toEqual(0);
+        done();
+      });
+    });
+  });
+
+  function serverIs31OrLater(done) {
+    // lazy way of checking the version number
+    // if server has been set we know it is at least 3.1
+    if (!serverMetadata) {
+      done();
+      return false;
+    }
+    return true;
+  }
+
+  function countNodes(label, propertyKey, propertyValue) {
+    return new Promise((resolve, reject) => {
+      session.run(`MATCH (n: ${label} {${propertyKey}: ${propertyValue}}) RETURN count(n) AS count`).then(result => {
+        resolve(result.records[0].get('count').toNumber());
+      }).catch(error => reject(error));
+    });
+  }
+
   function withQueryInTmpSession(driver, callback) {
     const tmpSession = driver.session();
     return tmpSession.run('RETURN 1').then(() => {
@@ -653,4 +823,9 @@ describe('session', () => {
     const idleConnections = connectionPool._pools[address];
     return idleConnections.length;
   }
+
+  function verifyBookmark(bookmark) {
+    expect(bookmark).toBeDefined();
+    expect(bookmark).not.toBeNull();
+  }
 });

test/v1/transaction.test.js

@@ -16,7 +16,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-import neo4j from "../../lib/v1";
+import neo4j from '../../lib/v1';
 
 describe('transaction', () => {
 
@@ -449,6 +449,42 @@ describe('transaction', () => {
     }).catch(console.log);
   });
 
+  it('should be open when neither committed nor rolled back', () => {
+    const tx = session.beginTransaction();
+    expect(tx.isOpen()).toBeTruthy();
+  });
+
+  it('should not be open after commit', done => {
+    const tx = session.beginTransaction();
+
+    tx.run('CREATE ()').then(() => {
+      tx.commit().then(() => {
+        expect(tx.isOpen()).toBeFalsy();
+        done();
+      });
+    });
+  });
+
+  it('should not be open after rollback', done => {
+    const tx = session.beginTransaction();
+
+    tx.run('CREATE ()').then(() => {
+      tx.rollback().then(() => {
+        expect(tx.isOpen()).toBeFalsy();
+        done();
+      });
+    });
+  });
+
+  it('should not be open after run error', done => {
+    const tx = session.beginTransaction();
+
+    tx.run('RETURN').catch(() => {
+      expect(tx.isOpen()).toBeFalsy();
+      done();
+    });
+  });
+
   function expectSyntaxError(error) {
     expect(error.code).toBe('Neo.ClientError.Statement.SyntaxError');
   }