Expose Bolt V3 features in the API

Bolt V3 allows additional metadata to be attached to BEGIN and RUN
messages. This makes it possible to expose a couple new features in
the API.

Auto-commit transactions executed via `Session#run()` now support
bookmarks. In previous protocol versions, there was no good place in
the RUN message to attach bookmarks. Auto-commit transactions will
now use bookmarks given in `Driver#session(bookmarkOrBookmarks)`
function and participate in causal chaining within a session.

Transactions functions, auto-commit, and explicit transactions now
accept a configuration object with transaction timeout and metadata.
Example of a configuration object:

```
{
    timeout: 3000, // 3 seconds
    metadata: {key1: 42, key2: '42'}
}
```

where timeout is specified in milliseconds and metadata is an object
containing valid Cypher types.

Transactions that execute longer than the configured timeout will be
terminated by the database. This functionality allows limiting
query/transaction execution time. Specified timeout overrides the
default timeout configured in the database using
`dbms.transaction.timeout` setting. Examples:

Specified transaction metadata will be attached to the executing
transaction and visible in the output of `dbms.listQueries` and
`dbms.listTransactions` procedures. It will also get logged to the
`query.log`. This functionality makes it easier to tag transactions
and is equivalent to `dbms.setTXMetaData` procedure.

Examples:

```
var driver = neo4j.driver(...);
var session = driver.session();

var txConfig = {
  timeout: 5000, // 5 seconds
  metadata: {
    type: 'My Query',
    application: 'My App neo4j#1',
    sequence_number: 42
  }
};

// transaction configuration for auto-commit transaction
session.run('RETURN $x', {x: 42}, txConfig)
       .then(...)
       .catch(...)

// transaction configuration for explicit transaction
var tx = session.beginTransaction(txConfig);
tx.run(...);

// transaction configuration for read transaction function
session.readTransaction(tx => tx.run('RETURN $x', {x: 42}), txConfig)
       .then(...)
       .catch(...)

// transaction configuration for write transaction function
session.writeTransaction(tx => tx.run('RETURN $x', {x: 42}), txConfig)
       .then(...)
       .catch(...)
```

Author: lutovich

src/v1/driver.js

@@ -192,7 +192,7 @@ class Driver {
   session(mode, bookmarkOrBookmarks) {
     const sessionMode = Driver._validateSessionMode(mode);
     const connectionProvider = this._getOrCreateConnectionProvider();
-    const bookmark = new Bookmark(bookmarkOrBookmarks);
+    const bookmark = bookmarkOrBookmarks ? new Bookmark(bookmarkOrBookmarks) : Bookmark.empty();
     return new Session(sessionMode, connectionProvider, bookmark, this._config);
   }
 

src/v1/internal/bolt-protocol-v1.js

@@ -18,6 +18,9 @@
  */
 import RequestMessage from './request-message';
 import * as v1 from './packstream-v1';
+import {newError} from '../error';
+import Bookmark from './bookmark';
+import TxConfig from './tx-config';
 
 export default class BoltProtocol {
 
@@ -72,9 +75,12 @@ export default class BoltProtocol {
   /**
    * Begin an explicit transaction.
    * @param {Bookmark} bookmark the bookmark.
+   * @param {TxConfig} txConfig the configuration.
    * @param {StreamObserver} observer the response observer.
    */
-  beginTransaction(bookmark, observer) {
+  beginTransaction(bookmark, txConfig, observer) {
+    assertTxConfigIsEmpty(txConfig, this._connection, observer);
+
     const runMessage = RequestMessage.run('BEGIN', bookmark.asBeginTransactionParameters());
     const pullAllMessage = RequestMessage.pullAll();
 
@@ -87,24 +93,29 @@ export default class BoltProtocol {
    * @param {StreamObserver} observer the response observer.
    */
   commitTransaction(observer) {
-    this.run('COMMIT', {}, observer);
+    this.run('COMMIT', {}, Bookmark.empty(), TxConfig.empty(), observer);
   }
 
   /**
    * Rollback the explicit transaction.
    * @param {StreamObserver} observer the response observer.
    */
   rollbackTransaction(observer) {
-    this.run('ROLLBACK', {}, observer);
+    this.run('ROLLBACK', {}, Bookmark.empty(), TxConfig.empty(), observer);
   }
 
   /**
    * Send a Cypher statement through the underlying connection.
    * @param {string} statement the cypher statement.
    * @param {object} parameters the statement parameters.
+   * @param {Bookmark} bookmark the bookmark.
+   * @param {TxConfig} txConfig the auto-commit transaction configuration.
    * @param {StreamObserver} observer the response observer.
    */
-  run(statement, parameters, observer) {
+  run(statement, parameters, bookmark, txConfig, observer) {
+    // bookmark is ignored in this version of the protocol
+    assertTxConfigIsEmpty(txConfig, this._connection, observer);
+
     const runMessage = RequestMessage.run(statement, parameters);
     const pullAllMessage = RequestMessage.pullAll();
 
@@ -129,3 +140,20 @@ export default class BoltProtocol {
     return new v1.Unpacker(disableLosslessIntegers);
   }
 }
+
+/**
+ * @param {TxConfig} txConfig the auto-commit transaction configuration.
+ * @param {Connection} connection the connection.
+ * @param {StreamObserver} observer the response observer.
+ */
+function assertTxConfigIsEmpty(txConfig, connection, observer) {
+  if (!txConfig.isEmpty()) {
+    const error = newError('Driver is connected to the database that does not support transaction configuration. ' +
+      'Please upgrade to neo4j 3.5.0 or later in order to use this functionality');
+
+    // unsupported API was used, consider this a fatal error for the current connection
+    connection._handleFatalError(error);
+    observer.onError(error);
+    throw error;
+  }
+}

src/v1/internal/bolt-protocol-v3.js

@@ -47,9 +47,9 @@ export default class BoltProtocol extends BoltProtocolV2 {
     this._connection.write(message, observer, true);
   }
 
-  beginTransaction(bookmark, observer) {
+  beginTransaction(bookmark, txConfig, observer) {
     prepareToHandleSingleResponse(observer);
-    const message = RequestMessage.begin(bookmark);
+    const message = RequestMessage.begin(bookmark, txConfig);
     this._connection.write(message, observer, true);
   }
 
@@ -65,9 +65,8 @@ export default class BoltProtocol extends BoltProtocolV2 {
     this._connection.write(message, observer, true);
   }
 
-  run(statement, parameters, observer) {
-    const metadata = {};
-    const runMessage = RequestMessage.runWithMetadata(statement, parameters, metadata);
+  run(statement, parameters, bookmark, txConfig, observer) {
+    const runMessage = RequestMessage.runWithMetadata(statement, parameters, bookmark, txConfig);
     const pullAllMessage = RequestMessage.pullAll();
 
     this._connection.write(runMessage, observer, false);

src/v1/internal/bookmark.js

@@ -36,6 +36,10 @@ export default class Bookmark {
     this._maxValue = maxBookmark(this._values);
   }
 
+  static empty() {
+    return EMPTY_BOOKMARK;
+  }
+
   /**
    * Check if the given bookmark is meaningful and can be send to the database.
    * @return {boolean} returns <code>true</code> bookmark has a value, <code>false</code> otherwise.
@@ -80,6 +84,8 @@ export default class Bookmark {
   }
 }
 
+const EMPTY_BOOKMARK = new Bookmark(null);
+
 /**
  * Converts given value to an array.
  * @param {string|string[]} [value=undefined] argument to convert.

src/v1/internal/connection.js

@@ -225,7 +225,6 @@ export default class Connection {
    * failing, and the connection getting ejected from the session pool.
    *
    * @param error an error object, forwarded to all current and future subscribers
-   * @protected
    */
   _handleFatalError(error) {
     this._isBroken = true;

src/v1/internal/request-message.js

@@ -88,10 +88,11 @@ export default class RequestMessage {
   /**
    * Create a new BEGIN message.
    * @param {Bookmark} bookmark the bookmark.
+   * @param {TxConfig} txConfig the configuration.
    * @return {RequestMessage} new BEGIN message.
    */
-  static begin(bookmark) {
-    const metadata = {bookmarks: bookmark.values()};
+  static begin(bookmark, txConfig) {
+    const metadata = buildTxMetadata(bookmark, txConfig);
     return new RequestMessage(BEGIN, [metadata], () => `BEGIN ${JSON.stringify(metadata)}`);
   }
 
@@ -115,15 +116,37 @@ export default class RequestMessage {
    * Create a new RUN message with additional metadata.
    * @param {string} statement the cypher statement.
    * @param {object} parameters the statement parameters.
-   * @param {object} metadata the additional metadata.
+   * @param {Bookmark} bookmark the bookmark.
+   * @param {TxConfig} txConfig the configuration.
    * @return {RequestMessage} new RUN message with additional metadata.
    */
-  static runWithMetadata(statement, parameters, metadata) {
+  static runWithMetadata(statement, parameters, bookmark, txConfig) {
+    const metadata = buildTxMetadata(bookmark, txConfig);
     return new RequestMessage(RUN, [statement, parameters, metadata],
       () => `RUN ${statement} ${JSON.stringify(parameters)} ${JSON.stringify(metadata)}`);
   }
 }
 
+/**
+ * Create an object that represent transaction metadata.
+ * @param {Bookmark} bookmark the bookmark.
+ * @param {TxConfig} txConfig the configuration.
+ * @return {object} a metadata object.
+ */
+function buildTxMetadata(bookmark, txConfig) {
+  const metadata = {};
+  if (!bookmark.isEmpty()) {
+    metadata['bookmarks'] = bookmark.values();
+  }
+  if (txConfig.timeout) {
+    metadata['tx_timeout'] = txConfig.timeout;
+  }
+  if (txConfig.metadata) {
+    metadata['tx_metadata'] = txConfig.metadata;
+  }
+  return metadata;
+}
+
 // constants for messages that never change
 const PULL_ALL_MESSAGE = new RequestMessage(PULL_ALL, [], () => 'PULL_ALL');
 const RESET_MESSAGE = new RequestMessage(RESET, [], () => 'RESET');

src/v1/internal/routing-util.js

@@ -20,6 +20,8 @@
 import {newError, PROTOCOL_ERROR, SERVICE_UNAVAILABLE} from '../error';
 import Integer, {int} from '../integer';
 import {ServerVersion, VERSION_3_2_0} from './server-version';
+import Bookmark from './bookmark';
+import TxConfig from './tx-config';
 
 const CALL_GET_SERVERS = 'CALL dbms.cluster.routing.getServers';
 const CALL_GET_ROUTING_TABLE = 'CALL dbms.cluster.routing.getRoutingTable($context)';
@@ -123,7 +125,7 @@ export default class RoutingUtil {
         params = {};
       }
 
-      connection.protocol().run(query, params, streamObserver);
+      connection.protocol().run(query, params, Bookmark.empty(), TxConfig.empty(), streamObserver);
     });
   }
 }

src/v1/internal/server-version.js

@@ -112,6 +112,7 @@ function compareInts(x, y) {
 const VERSION_3_1_0 = new ServerVersion(3, 1, 0);
 const VERSION_3_2_0 = new ServerVersion(3, 2, 0);
 const VERSION_3_4_0 = new ServerVersion(3, 4, 0);
+const VERSION_3_5_0 = new ServerVersion(3, 5, 0);
 const maxVer = Number.MAX_SAFE_INTEGER;
 const VERSION_IN_DEV = new ServerVersion(maxVer, maxVer, maxVer);
 
@@ -120,6 +121,7 @@ export {
   VERSION_3_1_0,
   VERSION_3_2_0,
   VERSION_3_4_0,
+  VERSION_3_5_0,
   VERSION_IN_DEV
 };
 

src/v1/internal/tx-config.js

@@ -0,0 +1,98 @@
+/**
+ * Copyright (c) 2002-2018 "Neo4j,"
+ * Neo4j Sweden AB [http://neo4j.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 * as util from './util';
+import {int} from '../integer';
+import {newError} from '../error';
+
+/**
+ * Internal holder of the transaction configuration.
+ * It performs input validation and value conversion for further serialization by the Bolt protocol layer.
+ * Users of the driver provide transaction configuration as regular objects `{timeout: 10, metadata: {key: 'value'}}`.
+ * Driver converts such objects to {@link TxConfig} immediately and uses converted values everywhere.
+ */
+export default class TxConfig {
+
+  /**
+   * @constructor
+   * @param {object} config the raw configuration object.
+   */
+  constructor(config) {
+    assertValidConfig(config);
+    this.timeout = extractTimeout(config);
+    this.metadata = extractMetadata(config);
+  }
+
+  /**
+   * Get an empty config object.
+   * @return {TxConfig} an empty config.
+   */
+  static empty() {
+    return EMPTY_CONFIG;
+  }
+
+  /**
+   * Check if this config object is empty. I.e. has no configuration values specified.
+   * @return {boolean} `true` if this object is empty, `false` otherwise.
+   */
+  isEmpty() {
+    return Object.values(this).every(value => value == null);
+  }
+}
+
+const EMPTY_CONFIG = new TxConfig({});
+
+/**
+ * @return {Integer|null}
+ */
+function extractTimeout(config) {
+  if (util.isObject(config) && (config.timeout || config.timeout === 0)) {
+    util.assertNumberOrInteger(config.timeout, 'Transaction timeout');
+    const timeout = int(config.timeout);
+    if (timeout.isZero()) {
+      throw newError('Transaction timeout should not be zero');
+    }
+    if (timeout.isNegative()) {
+      throw newError('Transaction timeout should not be negative');
+    }
+    return timeout;
+  }
+  return null;
+}
+
+/**
+ * @return {object|null}
+ */
+function extractMetadata(config) {
+  if (util.isObject(config) && config.metadata) {
+    const metadata = config.metadata;
+    util.assertObject(metadata);
+    if (Object.keys(metadata).length !== 0) {
+      // not an empty object
+      return metadata;
+    }
+  }
+  return null;
+}
+
+function assertValidConfig(config) {
+  if (config) {
+    util.assertObject(config, 'Transaction config');
+  }
+}

src/v1/internal/util.js

@@ -66,6 +66,13 @@ function validateStatementAndParameters(statement, parameters) {
   return {query, params};
 }
 
+function assertObject(obj, objName) {
+  if (!isObject(obj)) {
+    throw new TypeError(objName + ' expected to be an object but was: ' + JSON.stringify(obj));
+  }
+  return obj;
+}
+
 function assertString(obj, objName) {
   if (!isString(obj)) {
     throw new TypeError(objName + ' expected to be string but was: ' + JSON.stringify(obj));
@@ -118,7 +125,9 @@ function isString(str) {
 
 export {
   isEmptyObjectOrNull,
+  isObject,
   isString,
+  assertObject,
   assertString,
   assertNumber,
   assertNumberOrInteger,