Introduce transaction with retries API

This commit adds two new API functions:
 * `Session#readTransaction()`
 * `Session#writeTransaction()`

Both take a single function as input. This function takes a single argument of
type `Transaction` and returns a promise. It can be used to perform regular
async operations like query execution. Introduced functions will
commit/rollback transaction depending on the returned promise, so user code
does not need to call `Transaction#commit()` explicitly. They also perform
retries if given transaction fails with network errors
(`ServiceUnavaliable` or `SessionExpired`) or with transient errors
(`DeadlockDetected`, etc.). Retries are one with exponential backoff with
initial delay of 1 second and total cap of 30 seconds.

These API functions are useful to hide tolerable network problems and
transient errors for both single and causal cluster deployments.

Commit also fixes a problem in transaction error handling where `_onError`
executed a rollback but did not properly wait for the returned
promise to complete.

Author: lutovich

src/v1/internal/transaction-executor.js

@@ -0,0 +1,135 @@
+/**
+ * Copyright (c) 2002-2017 "Neo Technology,","
+ * Network Engine for Objects in Lund AB [http://neotechnology.com]
+ *
+ * This file is part of Neo4j.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import {newError, SERVICE_UNAVAILABLE, SESSION_EXPIRED} from '../error';
+
+const DEFAULT_MAX_RETRY_TIME_MS = 30 * 1000; // 30 seconds
+const DEFAULT_INITIAL_RETRY_DELAY_MS = 1000; // 1 seconds
+const DEFAULT_RETRY_DELAY_MULTIPLIER = 2.0;
+const DEFAULT_RETRY_DELAY_JITTER_FACTOR = 0.2;
+
+export default class TransactionExecutor {
+
+  constructor(maxRetryTimeMs, initialRetryDelayMs, multiplier, jitterFactor) {
+    this._maxRetryTimeMs = maxRetryTimeMs || DEFAULT_MAX_RETRY_TIME_MS;
+    this._initialRetryDelayMs = initialRetryDelayMs || DEFAULT_INITIAL_RETRY_DELAY_MS;
+    this._multiplier = multiplier || DEFAULT_RETRY_DELAY_MULTIPLIER;
+    this._jitterFactor = jitterFactor || DEFAULT_RETRY_DELAY_JITTER_FACTOR;
+
+    this._inFlightTimeoutIds = [];
+
+    this._verifyAfterConstruction();
+  }
+
+  execute(transactionCreator, transactionWork) {
+    return new Promise((resolve, reject) => {
+      this._executeTransactionInsidePromise(transactionCreator, transactionWork, resolve, reject);
+    }).catch(error => {
+      const retryStartTimeMs = Date.now();
+      const retryDelayMs = this._initialRetryDelayMs;
+      return this._retryTransactionPromise(transactionCreator, transactionWork, error, retryStartTimeMs, retryDelayMs);
+    });
+  }
+
+  close() {
+    // cancel all existing timeouts to prevent further retries
+    this._inFlightTimeoutIds.forEach(timeoutId => clearTimeout(timeoutId));
+    this._inFlightTimeoutIds = [];
+  }
+
+  _retryTransactionPromise(transactionCreator, transactionWork, error, retryStartTime, retryDelayMs) {
+    const elapsedTimeMs = Date.now() - retryStartTime;
+
+    if (elapsedTimeMs > this._maxRetryTimeMs || !TransactionExecutor._canRetryOn(error)) {
+      return Promise.reject(error);
+    }
+
+    return new Promise((resolve, reject) => {
+      const nextRetryTime = this._computeDelayWithJitter(retryDelayMs);
+      const timeoutId = setTimeout(() => {
+        // filter out this timeoutId when time has come and function is being executed
+        this._inFlightTimeoutIds = this._inFlightTimeoutIds.filter(id => id !== timeoutId);
+        this._executeTransactionInsidePromise(transactionCreator, transactionWork, resolve, reject);
+      }, nextRetryTime);
+      // add newly created timeoutId to the list of all in-flight timeouts
+      this._inFlightTimeoutIds.push(timeoutId);
+    }).catch(error => {
+      const nextRetryDelayMs = retryDelayMs * this._multiplier;
+      return this._retryTransactionPromise(transactionCreator, transactionWork, error, retryStartTime, nextRetryDelayMs);
+    });
+  }
+
+  _executeTransactionInsidePromise(transactionCreator, transactionWork, resolve, reject) {
+    try {
+      const tx = transactionCreator();
+      const transactionWorkResult = transactionWork(tx);
+
+      // user defined callback is supposed to return a promise, but it might not; so to protect against an
+      // incorrect API usage we wrap the returned value with a resolved promise; this is effectively a
+      // validation step without type checks
+      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
+          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);
+      });
+
+    } catch (error) {
+      reject(error);
+    }
+  }
+
+  _computeDelayWithJitter(delayMs) {
+    const jitter = (delayMs * this._jitterFactor);
+    const min = delayMs - jitter;
+    const max = delayMs + jitter;
+    return Math.random() * (max - min) + min;
+  }
+
+  static _canRetryOn(error) {
+    return error && error.code &&
+      (error.code === SERVICE_UNAVAILABLE ||
+      error.code === SESSION_EXPIRED ||
+      error.code.indexOf('TransientError') >= 0);
+  }
+
+  _verifyAfterConstruction() {
+    if (this._maxRetryTimeMs < 0) {
+      throw newError('Max retry time should be >= 0: ' + this._maxRetryTimeMs);
+    }
+    if (this._initialRetryDelayMs < 0) {
+      throw newError('Initial retry delay should >= 0: ' + this._initialRetryDelayMs);
+    }
+    if (this._multiplier < 1.0) {
+      throw newError('Multiplier should be >= 1.0: ' + this._multiplier);
+    }
+    if (this._jitterFactor < 0 || this._jitterFactor > 1) {
+      throw newError('Jitter factor should be in [0.0, 1.0]: ' + this._jitterFactor);
+    }
+  }
+};

src/v1/session.js

@@ -22,7 +22,8 @@ import Transaction from './transaction';
 import {newError} from './error';
 import {assertString} from './internal/util';
 import ConnectionHolder from './internal/connection-holder';
-import {READ, WRITE} from './driver';
+import Driver, {READ, WRITE} from './driver';
+import TransactionExecutor from './internal/transaction-executor';
 
 /**
   * A Session instance is used for handling the connection and
@@ -36,7 +37,7 @@ class Session {
    * @constructor
    * @param {string} mode the default access mode for this session.
    * @param {ConnectionProvider} connectionProvider - the connection provider to acquire connections from.
-   * @param {string} bookmark - the initial bookmark for this session.
+   * @param {string} [bookmark=undefined] - the initial bookmark for this session.
    */
   constructor(mode, connectionProvider, bookmark) {
     this._mode = mode;
@@ -45,6 +46,7 @@ class Session {
     this._open = true;
     this._hasTx = false;
     this._lastBookmark = bookmark;
+    this._transactionExecutor = new TransactionExecutor()
   }
 
   /**
@@ -92,21 +94,24 @@ class Session {
    * @returns {Transaction} - New Transaction
    */
   beginTransaction(bookmark) {
+    return this._beginTransaction(this._mode, bookmark);
+  }
+
+  _beginTransaction(accessMode, bookmark) {
     if (bookmark) {
       assertString(bookmark, 'Bookmark');
       this._updateBookmark(bookmark);
     }
 
     if (this._hasTx) {
-      throw newError("You cannot begin a transaction on a session with an "
-      + "open transaction; either run from within the transaction or use a "
-      + "different session.")
+      throw newError('You cannot begin a transaction on a session with an open transaction; ' +
+        'either run from within the transaction or use a different session.');
     }
 
-    this._hasTx = true;
-
-    const connectionHolder = this._connectionHolderWithMode(this._mode);
+    const mode = Driver._validateSessionMode(accessMode);
+    const connectionHolder = this._connectionHolderWithMode(mode);
     connectionHolder.initializeConnection();
+    this._hasTx = true;
 
     return new Transaction(connectionHolder, () => {
         this._hasTx = false;
@@ -118,6 +123,21 @@ class Session {
     return this._lastBookmark;
   }
 
+  readTransaction(transactionWork) {
+    return this._runTransaction(READ, transactionWork);
+  }
+
+  writeTransaction(transactionWork) {
+    return this._runTransaction(WRITE, transactionWork);
+  }
+
+  _runTransaction(accessMode, transactionWork) {
+    return this._transactionExecutor.execute(
+      () => this._beginTransaction(accessMode, this.lastBookmark()),
+      transactionWork
+    );
+  }
+
   _updateBookmark(newBookmark) {
     if (newBookmark) {
       this._lastBookmark = newBookmark;
@@ -132,6 +152,7 @@ class Session {
   close(callback = (() => null)) {
     if (this._open) {
       this._open = false;
+      this._transactionExecutor.close();
       this._readConnectionHolder.close().then(() => {
         this._writeConnectionHolder.close().then(() => {
           callback();

src/v1/transaction.js

@@ -104,14 +104,21 @@ class Transaction {
   }
 
   _onError() {
-    // rollback explicitly if tx.run failed, rollback
     if (this._state == _states.ACTIVE) {
-        this.rollback();
+      // 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
+      }).then(() => {
+        // after rollback attempt change this transaction's state to FAILED
+        this._state = _states.FAILED;
+      });
     } else {
-        // else just do the cleanup
-        this._onClose();
+      // error happened in in-active transaction, just to the cleanup and change state to FAILED
+      this._state = _states.FAILED;
+      this._onClose();
+      // no async actions needed - return resolved promise
+      return Promise.resolve();
     }
-    this._state = _states.FAILED;
   }
 }
 
@@ -126,9 +133,10 @@ class _TransactionStreamObserver extends StreamObserver {
 
   onError(error) {
     if (!this._hasFailed) {
-      this._tx._onError();
-      super.onError(error);
-      this._hasFailed = true;
+      this._tx._onError().then(() => {
+        super.onError(error);
+        this._hasFailed = true;
+      });
     }
   }
 

test/internal/timers-util.js

@@ -0,0 +1,86 @@
+/**
+ * Copyright (c) 2002-2017 "Neo Technology,","
+ * Network Engine for Objects in Lund AB [http://neotechnology.com]
+ *
+ * This file is part of Neo4j.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+class SetTimeoutMock {
+
+  constructor() {
+    this._clearState();
+  }
+
+  install() {
+    this._originalSetTimeout = global.setTimeout;
+    global.setTimeout = (code, delay) => {
+      if (!this._paused) {
+        code();
+        this.invocationDelays.push(delay);
+      }
+      return this._timeoutIdCounter++;
+    };
+
+    this._originalClearTimeout = global.clearTimeout;
+    global.clearTimeout = id => {
+      this.clearedTimeouts.push(id);
+    };
+
+    return this;
+  }
+
+  pause() {
+    this._paused = true;
+  }
+
+  uninstall() {
+    global.setTimeout = this._originalSetTimeout;
+    global.clearTimeout = this._originalClearTimeout;
+    this._clearState();
+  }
+
+  setTimeoutOriginal(code, delay) {
+    return this._originalSetTimeout.call(null, code, delay);
+  }
+
+  _clearState() {
+    this._originalSetTimeout = null;
+    this._originalClearTimeout = null;
+    this._paused = false;
+    this._timeoutIdCounter = 0;
+
+    this.invocationDelays = [];
+    this.clearedTimeouts = [];
+  }
+}
+
+export const setTimeoutMock = new SetTimeoutMock();
+
+export function hijackNextDateNowCall(newValue) {
+  const originalDate = global.Date;
+  global.Date = new FakeDate(originalDate, newValue);
+}
+
+class FakeDate {
+
+  constructor(originalDate, nextNowValue) {
+    this._originalDate = originalDate;
+    this._nextNowValue = nextNowValue;
+  }
+
+  now() {
+    global.Date = this._originalDate;
+    return this._nextNowValue;
+  }
+}