Add idempotency (#6748)

* added idempotency router and middleware

* added idempotency rules for routes classes, functions, jobs, installaions, users

* fixed typo

* ignore requests without header

* removed unused var

* enabled feature only for MongoDB

* changed code comment

* fixed inconsistend storage adapter specification

* Trigger notification

* Travis CI trigger

* Travis CI trigger

* Travis CI trigger

* rebuilt option definitions

* fixed incorrect import path

* added new request ID header to allowed headers

* fixed typescript typos

* add new system class to spec helper

* fixed typescript typos

* re-added postgres conn parameter

* removed postgres conn parameter

* fixed incorrect schema for index creation

* temporarily disabling index creation to fix postgres issue

* temporarily disabling index creation to fix postgres issue

* temporarily disabling index creation to fix postgres issue

* temporarily disabling index creation to fix postgres issue

* temporarily disabling index creation to fix postgres issue

* temporarily disabling index creation to fix postgres issue

* temporarily disabling index creation to fix postgres issue

* trying to fix postgres issue

* fixed incorrect auth when writing to _Idempotency

* trying to fix postgres issue

* Travis CI trigger

* added test cases

* removed number grouping

* fixed test description

* trying to fix postgres issue

* added Github readme docs

* added change log

* refactored tests; fixed some typos

* fixed test case

* fixed default TTL value

* Travis CI Trigger

* Travis CI Trigger

* Travis CI Trigger

* added test case to increase coverage

* Trigger Travis CI

* changed configuration syntax to use regex; added test cases

* removed unused vars

* removed IdempotencyRouter

* Trigger Travis CI

* updated docs

* updated docs

* updated docs

* updated docs

* update docs

* Trigger Travis CI

* fixed coverage

* removed code comments

src/Adapters/Storage/Mongo/MongoStorageAdapter.js

@@ -692,7 +692,7 @@ export class MongoStorageAdapter implements StorageAdapter {
     fieldNames: string[],
     indexName: ?string,
     caseInsensitive: boolean = false,
-    indexType: any = 1
+    options?: Object = {},
   ): Promise<any> {
     schema = convertParseSchemaToMongoSchema(schema);
     const indexCreationRequest = {};
@@ -700,11 +700,12 @@ export class MongoStorageAdapter implements StorageAdapter {
       transformKey(className, fieldName, schema)
     );
     mongoFieldNames.forEach((fieldName) => {
-      indexCreationRequest[fieldName] = indexType;
+      indexCreationRequest[fieldName] = options.indexType !== undefined ? options.indexType : 1;
     });
 
     const defaultOptions: Object = { background: true, sparse: true };
     const indexNameOptions: Object = indexName ? { name: indexName } : {};
+    const ttlOptions: Object = options.ttl !== undefined ? { expireAfterSeconds: options.ttl } : {};
     const caseInsensitiveOptions: Object = caseInsensitive
       ? { collation: MongoCollection.caseInsensitiveCollation() }
       : {};
@@ -712,6 +713,7 @@ export class MongoStorageAdapter implements StorageAdapter {
       ...defaultOptions,
       ...caseInsensitiveOptions,
       ...indexNameOptions,
+      ...ttlOptions,
     };
 
     return this._adaptiveCollection(className)

src/Adapters/Storage/Postgres/PostgresStorageAdapter.js

@@ -1209,6 +1209,7 @@ export class PostgresStorageAdapter implements StorageAdapter {
             '_GlobalConfig',
             '_GraphQLConfig',
             '_Audience',
+            '_Idempotency',
             ...results.map((result) => result.className),
             ...joins,
           ];
@@ -2576,9 +2577,9 @@ export class PostgresStorageAdapter implements StorageAdapter {
     fieldNames: string[],
     indexName: ?string,
     caseInsensitive: boolean = false,
-    conn: ?any = null
+    options?: Object = {},
   ): Promise<any> {
-    conn = conn != null ? conn : this._client;
+    const conn = options.conn !== undefined ? options.conn : this._client;
     const defaultIndexName = `parse_default_${fieldNames.sort().join('_')}`;
     const indexNameOptions: Object =
       indexName != null ? { name: indexName } : { name: defaultIndexName };

src/Adapters/Storage/StorageAdapter.js

@@ -93,7 +93,7 @@ export interface StorageAdapter {
     fieldNames: string[],
     indexName?: string,
     caseSensitive?: boolean,
-    indexType?: any
+    options?: Object,
   ): Promise<any>;
   ensureUniqueness(
     className: string,

src/Config.js

@@ -6,6 +6,7 @@ import AppCache from './cache';
 import SchemaCache from './Controllers/SchemaCache';
 import DatabaseController from './Controllers/DatabaseController';
 import net from 'net';
+import { IdempotencyOptions } from './Options/Definitions';
 
 function removeTrailingSlash(str) {
   if (!str) {
@@ -73,6 +74,7 @@ export class Config {
     masterKey,
     readOnlyMasterKey,
     allowHeaders,
+    idempotencyOptions,
   }) {
     if (masterKey === readOnlyMasterKey) {
       throw new Error('masterKey and readOnlyMasterKey should be different');
@@ -104,14 +106,27 @@ export class Config {
         throw 'publicServerURL should be a valid HTTPS URL starting with https://';
       }
     }
-
     this.validateSessionConfiguration(sessionLength, expireInactiveSessions);
-
     this.validateMasterKeyIps(masterKeyIps);
-
     this.validateMaxLimit(maxLimit);
-
     this.validateAllowHeaders(allowHeaders);
+    this.validateIdempotencyOptions(idempotencyOptions);
+  }
+
+  static validateIdempotencyOptions(idempotencyOptions) {
+    if (!idempotencyOptions) { return; }
+    if (idempotencyOptions.ttl === undefined) {
+      idempotencyOptions.ttl = IdempotencyOptions.ttl.default;
+    } else if (!isNaN(idempotencyOptions.ttl) && idempotencyOptions.ttl <= 0) {
+      throw 'idempotency TTL value must be greater than 0 seconds';
+    } else if (isNaN(idempotencyOptions.ttl)) {
+      throw 'idempotency TTL value must be a number';
+    }
+    if (!idempotencyOptions.paths) {
+      idempotencyOptions.paths = IdempotencyOptions.paths.default;
+    } else if (!(idempotencyOptions.paths instanceof Array)) {
+      throw 'idempotency paths must be of an array of strings';
+    }
   }
 
   static validateAccountLockoutPolicy(accountLockout) {

src/Controllers/DatabaseController.js

@@ -244,6 +244,7 @@ const filterSensitiveData = (
 };
 
 import type { LoadSchemaOptions } from './types';
+import MongoStorageAdapter from '../Adapters/Storage/Mongo/MongoStorageAdapter';
 
 // Runs an update on the database.
 // Returns a promise for an object with the new values for field
@@ -1736,6 +1737,12 @@ class DatabaseController {
         ...SchemaController.defaultColumns._Role,
       },
     };
+    const requiredIdempotencyFields = {
+      fields: {
+        ...SchemaController.defaultColumns._Default,
+        ...SchemaController.defaultColumns._Idempotency,
+      },
+    };
 
     const userClassPromise = this.loadSchema().then(schema =>
       schema.enforceClassExists('_User')
@@ -1743,6 +1750,9 @@ class DatabaseController {
     const roleClassPromise = this.loadSchema().then(schema =>
       schema.enforceClassExists('_Role')
     );
+    const idempotencyClassPromise = this.adapter instanceof MongoStorageAdapter
+      ? this.loadSchema().then((schema) => schema.enforceClassExists('_Idempotency'))
+      : Promise.resolve();
 
     const usernameUniqueness = userClassPromise
       .then(() =>
@@ -1807,6 +1817,43 @@ class DatabaseController {
         throw error;
       });
 
+    const idempotencyRequestIdIndex = this.adapter instanceof MongoStorageAdapter
+      ? idempotencyClassPromise
+        .then(() =>
+          this.adapter.ensureUniqueness(
+            '_Idempotency',
+            requiredIdempotencyFields,
+            ['reqId']
+          ))
+        .catch((error) => {
+          logger.warn(
+            'Unable to ensure uniqueness for idempotency request ID: ',
+            error
+          );
+          throw error;
+        })
+      : Promise.resolve();
+
+    const idempotencyExpireIndex = this.adapter instanceof MongoStorageAdapter
+      ? idempotencyClassPromise
+        .then(() =>
+          this.adapter.ensureIndex(
+            '_Idempotency',
+            requiredIdempotencyFields,
+            ['expire'],
+            'ttl',
+            false,
+            { ttl: 0 },
+          ))
+        .catch((error) => {
+          logger.warn(
+            'Unable to create TTL index for idempotency expire date: ',
+            error
+          );
+          throw error;
+        })
+      : Promise.resolve();
+
     const indexPromise = this.adapter.updateSchemaWithIndexes();
 
     // Create tables for volatile classes
@@ -1819,6 +1866,8 @@ class DatabaseController {
       emailUniqueness,
       emailCaseInsensitiveIndex,
       roleUniqueness,
+      idempotencyRequestIdIndex,
+      idempotencyExpireIndex,
       adapterInit,
       indexPromise,
     ]);

src/Controllers/SchemaController.js

@@ -144,6 +144,10 @@ const defaultColumns: { [string]: SchemaFields } = Object.freeze({
     lastUsed: { type: 'Date' },
     timesUsed: { type: 'Number' },
   },
+  _Idempotency: {
+    reqId: { type: 'String' },
+    expire: { type: 'Date' },
+  }
 });
 
 const requiredColumns = Object.freeze({
@@ -161,6 +165,7 @@ const systemClasses = Object.freeze([
   '_JobStatus',
   '_JobSchedule',
   '_Audience',
+  '_Idempotency'
 ]);
 
 const volatileClasses = Object.freeze([
@@ -171,6 +176,7 @@ const volatileClasses = Object.freeze([
   '_GraphQLConfig',
   '_JobSchedule',
   '_Audience',
+  '_Idempotency'
 ]);
 
 // Anything that start with role
@@ -660,6 +666,13 @@ const _AudienceSchema = convertSchemaToAdapterSchema(
     classLevelPermissions: {},
   })
 );
+const _IdempotencySchema = convertSchemaToAdapterSchema(
+  injectDefaultSchema({
+    className: '_Idempotency',
+    fields: defaultColumns._Idempotency,
+    classLevelPermissions: {},
+  })
+);
 const VolatileClassesSchemas = [
   _HooksSchema,
   _JobStatusSchema,
@@ -668,6 +681,7 @@ const VolatileClassesSchemas = [
   _GlobalConfigSchema,
   _GraphQLConfigSchema,
   _AudienceSchema,
+  _IdempotencySchema
 ];
 
 const dbTypeMatchesObjectType = (

src/Options/docs.js

@@ -33,6 +33,7 @@
  * @property {String} graphQLPath Mount path for the GraphQL endpoint, defaults to /graphql
  * @property {String} graphQLSchema Full path to your GraphQL custom schema.graphql file
  * @property {String} host The host to serve ParseServer on, defaults to 0.0.0.0
+ * @property {IdempotencyOptions} idempotencyOptions Options for request idempotency to deduplicate identical requests that may be caused by network issues. Caution, this is an experimental feature that may not be appropriate for production.
  * @property {String} javascriptKey Key for the Javascript SDK
  * @property {Boolean} jsonLogs Log as structured JSON objects
  * @property {LiveQueryOptions} liveQuery parse-server's LiveQuery configuration object
@@ -111,3 +112,10 @@
  * @property {Number} websocketTimeout Number of milliseconds between ping/pong frames. The WebSocket server sends ping/pong frames to the clients to keep the WebSocket alive. This value defines the interval of the ping/pong frame from the server to clients, defaults to 10 * 1000 ms (10 s).
  * @property {Adapter<WSSAdapter>} wssAdapter Adapter module for the WebSocketServer
  */
+
+/**
+ * @interface IdempotencyOptions
+ * @property {String[]} paths An array of paths for which the feature should be enabled. The mount path must not be included, for example instead of `/parse/functions/myFunction` specifiy `functions/myFunction`. The entries are interpreted as regular expression, for example `functions/.*` matches all functions, `jobs/.*` matches all jobs, `classes/.*` matches all classes, `.*` matches all paths.
+ * @property {Number} ttl The duration in seconds after which a request record is discarded from the database, defaults to 300s.
+ */
+

src/Options/index.js

@@ -188,6 +188,10 @@ export interface ParseServerOptions {
   startLiveQueryServer: ?boolean;
   /* Live query server configuration options (will start the liveQuery server) */
   liveQueryServerOptions: ?LiveQueryServerOptions;
+  /* Options for request idempotency to deduplicate identical requests that may be caused by network issues. Caution, this is an experimental feature that may not be appropriate for production.
+  :ENV: PARSE_SERVER_EXPERIMENTAL_IDEMPOTENCY_OPTIONS
+  :DEFAULT: false */
+  idempotencyOptions: ?IdempotencyOptions;
   /* Full path to your GraphQL custom schema.graphql file */
   graphQLSchema: ?string;
   /* Mounts the GraphQL endpoint
@@ -272,3 +276,12 @@ export interface LiveQueryServerOptions {
   /* Adapter module for the WebSocketServer */
   wssAdapter: ?Adapter<WSSAdapter>;
 }
+
+export interface IdempotencyOptions {
+  /* An array of paths for which the feature should be enabled. The mount path must not be included, for example instead of `/parse/functions/myFunction` specifiy `functions/myFunction`. The entries are interpreted as regular expression, for example `functions/.*` matches all functions, `jobs/.*` matches all jobs, `classes/.*` matches all classes, `.*` matches all paths.
+  :DEFAULT: [] */
+  paths: ?(string[]);
+  /* The duration in seconds after which a request record is discarded from the database, defaults to 300s.
+  :DEFAULT: 300 */
+  ttl: ?number;
+}

src/Routers/ClassesRouter.js

@@ -2,6 +2,7 @@ import PromiseRouter from '../PromiseRouter';
 import rest from '../rest';
 import _ from 'lodash';
 import Parse from 'parse/node';
+import { promiseEnsureIdempotency } from '../middlewares';
 
 const ALLOWED_GET_QUERY_KEYS = [
   'keys',
@@ -247,10 +248,10 @@ export class ClassesRouter extends PromiseRouter {
     this.route('GET', '/classes/:className/:objectId', req => {
       return this.handleGet(req);
     });
-    this.route('POST', '/classes/:className', req => {
+    this.route('POST', '/classes/:className', promiseEnsureIdempotency, req => {
       return this.handleCreate(req);
     });
-    this.route('PUT', '/classes/:className/:objectId', req => {
+    this.route('PUT', '/classes/:className/:objectId', promiseEnsureIdempotency, req => {
       return this.handleUpdate(req);
     });
     this.route('DELETE', '/classes/:className/:objectId', req => {

src/Routers/FunctionsRouter.js

@@ -4,7 +4,7 @@ var Parse = require('parse/node').Parse,
   triggers = require('../triggers');
 
 import PromiseRouter from '../PromiseRouter';
-import { promiseEnforceMasterKeyAccess } from '../middlewares';
+import { promiseEnforceMasterKeyAccess, promiseEnsureIdempotency } from '../middlewares';
 import { jobStatusHandler } from '../StatusHandler';
 import _ from 'lodash';
 import { logger } from '../logger';
@@ -34,11 +34,13 @@ export class FunctionsRouter extends PromiseRouter {
     this.route(
       'POST',
       '/functions/:functionName',
+      promiseEnsureIdempotency,
       FunctionsRouter.handleCloudFunction
     );
     this.route(
       'POST',
       '/jobs/:jobName',
+      promiseEnsureIdempotency,
       promiseEnforceMasterKeyAccess,
       function (req) {
         return FunctionsRouter.handleCloudJob(req);

src/Routers/InstallationsRouter.js

@@ -2,6 +2,7 @@
 
 import ClassesRouter from './ClassesRouter';
 import rest from '../rest';
+import { promiseEnsureIdempotency } from '../middlewares';
 
 export class InstallationsRouter extends ClassesRouter {
   className() {
@@ -36,10 +37,10 @@ export class InstallationsRouter extends ClassesRouter {
     this.route('GET', '/installations/:objectId', req => {
       return this.handleGet(req);
     });
-    this.route('POST', '/installations', req => {
+    this.route('POST', '/installations', promiseEnsureIdempotency, req => {
       return this.handleCreate(req);
     });
-    this.route('PUT', '/installations/:objectId', req => {
+    this.route('PUT', '/installations/:objectId', promiseEnsureIdempotency, req => {
       return this.handleUpdate(req);
     });
     this.route('DELETE', '/installations/:objectId', req => {

src/Routers/UsersRouter.js

@@ -8,6 +8,7 @@ import rest from '../rest';
 import Auth from '../Auth';
 import passwordCrypto from '../password';
 import { maybeRunTrigger, Types as TriggerTypes } from '../triggers';
+import { promiseEnsureIdempotency } from '../middlewares';
 
 export class UsersRouter extends ClassesRouter {
   className() {
@@ -445,7 +446,7 @@ export class UsersRouter extends ClassesRouter {
     this.route('GET', '/users', req => {
       return this.handleFind(req);
     });
-    this.route('POST', '/users', req => {
+    this.route('POST', '/users', promiseEnsureIdempotency, req => {
       return this.handleCreate(req);
     });
     this.route('GET', '/users/me', req => {
@@ -454,7 +455,7 @@ export class UsersRouter extends ClassesRouter {
     this.route('GET', '/users/:objectId', req => {
       return this.handleGet(req);
     });
-    this.route('PUT', '/users/:objectId', req => {
+    this.route('PUT', '/users/:objectId', promiseEnsureIdempotency, req => {
       return this.handleUpdate(req);
     });
     this.route('DELETE', '/users/:objectId', req => {

src/middlewares.js

@@ -4,9 +4,11 @@ import auth from './Auth';
 import Config from './Config';
 import ClientSDK from './ClientSDK';
 import defaultLogger from './logger';
+import rest from './rest';
+import MongoStorageAdapter from './Adapters/Storage/Mongo/MongoStorageAdapter';
 
 export const DEFAULT_ALLOWED_HEADERS =
-  'X-Parse-Master-Key, X-Parse-REST-API-Key, X-Parse-Javascript-Key, X-Parse-Application-Id, X-Parse-Client-Version, X-Parse-Session-Token, X-Requested-With, X-Parse-Revocable-Session, Content-Type, Pragma, Cache-Control';
+  'X-Parse-Master-Key, X-Parse-REST-API-Key, X-Parse-Javascript-Key, X-Parse-Application-Id, X-Parse-Client-Version, X-Parse-Session-Token, X-Requested-With, X-Parse-Revocable-Session, X-Parse-Request-Id, Content-Type, Pragma, Cache-Control';
 
 const getMountForRequest = function (req) {
   const mountPathLength = req.originalUrl.length - req.url.length;
@@ -406,6 +408,52 @@ export function promiseEnforceMasterKeyAccess(request) {
   return Promise.resolve();
 }
 
+/**
+ * Deduplicates a request to ensure idempotency. Duplicates are determined by the request ID
+ * in the request header. If a request has no request ID, it is executed anyway.
+ * @param {*} req The request to evaluate.
+ * @returns Promise<{}>
+ */
+export function promiseEnsureIdempotency(req) {
+  // Enable feature only for MongoDB
+  if (!(req.config.database.adapter instanceof MongoStorageAdapter)) { return Promise.resolve(); }
+  // Get parameters
+  const config = req.config;
+  const requestId = ((req || {}).headers || {})["x-parse-request-id"];
+  const { paths, ttl } = config.idempotencyOptions;
+  if (!requestId || !config.idempotencyOptions) { return Promise.resolve(); }
+  // Request path may contain trailing slashes, depending on the original request, so remove
+  // leading and trailing slashes to make it easier to specify paths in the configuration
+  const reqPath = req.path.replace(/^\/|\/$/, '');
+  // Determine whether idempotency is enabled for current request path
+  let match = false;
+  for (const path of paths) {
+    // Assume one wants a path to always match from the beginning to prevent any mistakes
+    const regex = new RegExp(path.charAt(0) === '^' ? path : '^' + path);
+    if (reqPath.match(regex)) {
+      match = true;
+      break;
+    }
+  }
+  if (!match) { return Promise.resolve(); }
+  // Try to store request
+  const expiryDate = new Date(new Date().setSeconds(new Date().getSeconds() + ttl));
+  return rest.create(
+    config,
+    auth.master(config),
+    '_Idempotency',
+    { reqId: requestId, expire: Parse._encode(expiryDate) }
+  ).catch (e => {
+    if (e.code == Parse.Error.DUPLICATE_VALUE) {
+      throw new Parse.Error(
+        Parse.Error.DUPLICATE_REQUEST,
+        'Duplicate request'
+      );
+    }
+    throw e;
+  });
+}
+
 function invalidRequest(req, res) {
   res.status(403);
   res.end('{"error":"unauthorized"}');

src/rest.js

@@ -284,6 +284,7 @@ const classesWithMasterOnlyAccess = [
   '_Hooks',
   '_GlobalConfig',
   '_JobSchedule',
+  '_Idempotency',
 ];
 // Disallowing access to the _Role collection except by master key
 function enforceRoleSecurity(method, className, auth) {