Rollback transaction when HTTP session is closed

HTTP driver uses transactional Cypher endpoint instead of Bolt. It used
to simply submit queries via `POST db/data/transaction/commit` in
auto-commit mode. This made it impossible to cancel such queries because
transaction id was not known. Canceling is needed in `Session#close()`
to terminate all ongoing queries and transactions. `Session#close()`
is used by neo4j-browser to terminate running queries.

This commit makes `Session#close()` terminate all ongoing transactions
and thus cancel all running queries. `Session#run()` now does not submit
query directly via `POST db/data/transaction/commit`. Instead, it first
starts a transaction via `POST db/data/transaction` and gets id of the
newly started transaction. Then a query is submitted via
`POST db/data/transaction/XXX` where `XXX` is the transaction id. After
a successful query, separate commit request is issued via
`POST db/data/transaction/XXX/commit`. After a failed query, rollback
is issued via `DELETE db/data/transaction/XXX`.

Transaction id is thus known during the query execution and can be used
by `Session#close()` to rollback the transaction via
`DELETE db/data/transaction/XXX`.

Author: lutovich

src/v1/internal/http/http-request-runner.js

@@ -0,0 +1,194 @@
+/**
+ * Copyright (c) 2002-2018 "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 StreamObserver from '../stream-observer';
+import HttpResponseConverter from './http-response-converter';
+import {Neo4jError, SERVICE_UNAVAILABLE} from '../../error';
+
+export default class HttpRequestRunner {
+
+  constructor(url, authToken) {
+    this._url = url;
+    this._authToken = authToken;
+    this._converter = new HttpResponseConverter();
+  }
+
+  /**
+   * Send a HTTP request to begin a transaction.
+   * @return {Promise<number>} promise resolved with the transaction id or rejected with an error.
+   */
+  beginTransaction() {
+    const url = beginTransactionUrl(this._url);
+    return sendRequest('POST', url, null, this._authToken).then(responseJson => {
+      const neo4jError = this._converter.extractError(responseJson);
+      if (neo4jError) {
+        throw neo4jError;
+      }
+      return this._converter.extractTransactionId(responseJson);
+    });
+  }
+
+  /**
+   * Send a HTTP request to commit a transaction.
+   * @param {number} transactionId id of the transaction to commit.
+   * @return {Promise<void>} promise resolved if transaction got committed or rejected when commit failed.
+   */
+  commitTransaction(transactionId) {
+    const url = commitTransactionUrl(this._url, transactionId);
+    return sendRequest('POST', url, null, this._authToken).then(responseJson => {
+      const neo4jError = this._converter.extractError(responseJson);
+      if (neo4jError) {
+        throw neo4jError;
+      }
+    });
+  }
+
+  /**
+   * Send a HTTP request to rollback a transaction.
+   * @param {number} transactionId id of the transaction to rollback.
+   * @return {Promise<void>} promise resolved if transaction got rolled back or rejected when rollback failed.
+   */
+  rollbackTransaction(transactionId) {
+    const url = transactionUrl(this._url, transactionId);
+    return sendRequest('DELETE', url, null, this._authToken).then(responseJson => {
+      const neo4jError = this._converter.extractError(responseJson);
+      if (neo4jError) {
+        throw neo4jError;
+      }
+    });
+  }
+
+  /**
+   * Send a HTTP request to execute a query in a transaction with the given id.
+   * @param {number} transactionId the transaction id.
+   * @param {string} statement the cypher query.
+   * @param {object} parameters the cypher query parameters.
+   * @return {Promise<StreamObserver>} a promise resolved with {@link StreamObserver} containing either records or error.
+   */
+  runQuery(transactionId, statement, parameters) {
+    const streamObserver = new StreamObserver();
+    const url = transactionUrl(this._url, transactionId);
+    const body = createStatementJson(statement, parameters, this._converter, streamObserver);
+    if (!body) {
+      // unable to encode given statement and parameters, return a failed stream observer
+      return Promise.resolve(streamObserver);
+    }
+
+    return sendRequest('POST', url, body, this._authToken).then(responseJson => {
+      processResponseJson(responseJson, this._converter, streamObserver);
+    }).catch(error => {
+      streamObserver.onError(error);
+    }).then(() => {
+      return streamObserver;
+    });
+  }
+}
+
+function sendRequest(method, url, bodyString, authToken) {
+  try {
+    const options = {
+      method: method,
+      headers: createHttpHeaders(authToken),
+      body: bodyString
+    };
+
+    return new Promise((resolve, reject) => {
+      fetch(url, options)
+        .then(response => response.json())
+        .then(responseJson => resolve(responseJson))
+        .catch(error => reject(new Neo4jError(error.message, SERVICE_UNAVAILABLE)));
+    });
+  } catch (e) {
+    return Promise.reject(e);
+  }
+}
+
+function createHttpHeaders(authToken) {
+  const headers = new Headers();
+  headers.append('Accept', 'application/json; charset=UTF-8');
+  headers.append('Content-Type', 'application/json');
+  headers.append('Authorization', 'Basic ' + btoa(authToken.principal + ':' + authToken.credentials));
+  return headers;
+}
+
+function createStatementJson(statement, parameters, converter, streamObserver) {
+  try {
+    return createStatementJsonOrThrow(statement, parameters, converter);
+  } catch (e) {
+    streamObserver.onError(e);
+    return null;
+  }
+}
+
+function createStatementJsonOrThrow(statement, parameters, converter) {
+  const encodedParameters = converter.encodeStatementParameters(parameters);
+  return JSON.stringify({
+    statements: [{
+      statement: statement,
+      parameters: encodedParameters,
+      resultDataContents: ['row', 'graph'],
+      includeStats: true
+    }]
+  });
+}
+
+function processResponseJson(responseJson, converter, streamObserver) {
+  if (!responseJson) {
+    // request failed and there is no response
+    return;
+  }
+
+  try {
+    processResponseJsonOrThrow(responseJson, converter, streamObserver);
+  } catch (e) {
+    streamObserver.onError(e);
+  }
+}
+
+function processResponseJsonOrThrow(responseJson, converter, streamObserver) {
+  const neo4jError = converter.extractError(responseJson);
+  if (neo4jError) {
+    streamObserver.onError(neo4jError);
+  } else {
+    const recordMetadata = converter.extractRecordMetadata(responseJson);
+    streamObserver.onCompleted(recordMetadata);
+
+    const rawRecords = converter.extractRawRecords(responseJson);
+    rawRecords.forEach(rawRecord => streamObserver.onNext(rawRecord));
+
+    const statementMetadata = converter.extractStatementMetadata(responseJson);
+    streamObserver.onCompleted(statementMetadata);
+  }
+}
+
+function beginTransactionUrl(baseUrl) {
+  return createUrl(baseUrl, '/db/data/transaction');
+}
+
+function commitTransactionUrl(baseUrl, transactionId) {
+  return transactionUrl(baseUrl, transactionId) + '/commit';
+}
+
+function transactionUrl(baseUrl, transactionId) {
+  return beginTransactionUrl(baseUrl) + '/' + transactionId;
+}
+
+function createUrl(baseUrl, path) {
+  return `${baseUrl.scheme}://${baseUrl.host}:${baseUrl.port}${path}`;
+}

src/v1/internal/http/http-data-converter.js renamed to src/v1/internal/http/http-response-converter.js

@@ -19,25 +19,16 @@
 
 import {isInt} from '../../integer';
 import {Node, Path, PathSegment, Relationship} from '../../graph-types';
-import {Neo4jError, SERVICE_UNAVAILABLE} from '../../error';
+import {Neo4jError} from '../../error';
 
 const CREDENTIALS_EXPIRED_CODE = 'Neo.ClientError.Security.CredentialsExpired';
 
-export default class HttpDataConverter {
+export default class HttpResponseConverter {
 
   encodeStatementParameters(parameters) {
     return encodeQueryParameters(parameters);
   }
 
-  /**
-   * Convert network error to a {@link Neo4jError}.
-   * @param {object} error the error to convert.
-   * @return {Neo4jError} new driver friendly error.
-   */
-  convertNetworkError(error) {
-    return new Neo4jError(error.message, SERVICE_UNAVAILABLE);
-  }
-
   /**
    * Attempts to extract error from transactional cypher endpoint response and convert it to {@link Neo4jError}.
    * @param {object} response the response.
@@ -59,6 +50,25 @@ export default class HttpDataConverter {
     return null;
   }
 
+  /**
+   * Extracts transaction id from the db/data/transaction endpoint response.
+   * @param {object} response the response.
+   * @return {number} the transaction id.
+   */
+  extractTransactionId(response) {
+    const commitUrl = response.commit;
+    if (commitUrl) {
+      // extract id 42 from commit url like 'http://localhost:7474/db/data/transaction/42/commit'
+      const url = commitUrl.replace('/commit', '');
+      const transactionIdString = url.substring(url.lastIndexOf('/') + 1);
+      const transactionId = parseInt(transactionIdString, 10);
+      if (transactionId || transactionId === 0) {
+        return transactionId;
+      }
+    }
+    throw new Neo4jError(`Unable to extract transaction id from the response JSON: ${JSON.stringify(response)}`);
+  }
+
   /**
    * Extracts record metadata (array of column names) from transactional cypher endpoint response.
    * @param {object} response the response.

src/v1/internal/http/http-session.js

@@ -21,13 +21,17 @@ import {WRITE} from '../../driver';
 import Session from '../../session';
 import {assertCypherStatement} from '../util';
 import {Neo4jError} from '../../error';
-import HttpStatementRunner from './http-statement-runner';
+import HttpRequestRunner from './http-request-runner';
+import {EMPTY_CONNECTION_HOLDER} from '../connection-holder';
+import Result from '../../result';
 
 export default class HttpSession extends Session {
 
   constructor(url, authToken, config) {
     super(WRITE, null, null, config);
-    this._statementRunner = new HttpStatementRunner(url, authToken);
+    this._ongoingTransactionIds = [];
+    this._serverInfoSupplier = createServerInfoSupplier(url);
+    this._requestRunner = new HttpRequestRunner(url, authToken);
   }
 
   run(statement, parameters = {}) {
@@ -37,7 +41,21 @@ export default class HttpSession extends Session {
     }
     assertCypherStatement(statement);
 
-    return this._statementRunner.run(statement, parameters);
+    return this._requestRunner.beginTransaction().then(transactionId => {
+      this._ongoingTransactionIds.push(transactionId);
+      const queryPromise = this._requestRunner.runQuery(transactionId, statement, parameters);
+
+      return queryPromise.then(streamObserver => {
+        if (streamObserver.hasFailed()) {
+          return rollbackTransactionAfterQueryFailure(transactionId, streamObserver, this._requestRunner);
+        } else {
+          return commitTransactionAfterQuerySuccess(transactionId, streamObserver, this._requestRunner);
+        }
+      }).then(streamObserver => {
+        this._ongoingTransactionIds = this._ongoingTransactionIds.filter(id => id !== transactionId);
+        return new Result(streamObserver, statement, parameters, this._serverInfoSupplier, EMPTY_CONNECTION_HOLDER);
+      });
+    });
   }
 
   beginTransaction() {
@@ -57,10 +75,38 @@ export default class HttpSession extends Session {
   }
 
   close(callback = (() => null)) {
-    callback();
+    const rollbackAllOngoingTransactions = this._ongoingTransactionIds.map(transactionId =>
+      rollbackTransactionSilently(transactionId, this._requestRunner)
+    );
+
+    Promise.all(rollbackAllOngoingTransactions).then(() => callback());
   }
 }
 
+function rollbackTransactionAfterQueryFailure(transactionId, streamObserver, requestRunner) {
+  return rollbackTransactionSilently(transactionId, requestRunner).then(() => streamObserver);
+}
+
+function commitTransactionAfterQuerySuccess(transactionId, streamObserver, requestRunner) {
+  return requestRunner.commitTransaction(transactionId).catch(error => {
+    streamObserver.onError(error);
+  }).then(() => {
+    return streamObserver;
+  });
+}
+
+function rollbackTransactionSilently(transactionId, requestRunner) {
+  return requestRunner.rollbackTransaction(transactionId).catch(error => {
+    // ignore all rollback errors
+  });
+}
+
+function createServerInfoSupplier(url) {
+  return () => {
+    return {server: {address: url.hostAndPort}};
+  };
+}
+
 function throwTransactionsNotSupported() {
   throw new Neo4jError('Experimental HTTP driver does not support transactions');
 }