Add KeyPromiseQueue to Push and Job StatusHandlers (#7267)

* Add KeyPromiseQueue to Push and Job StatusHandlers

* Update CHANGELOG.md

* Update CHANGELOG.md

CHANGELOG.md

@@ -116,6 +116,7 @@ ___
 - Test Parse Server continuously against all relevant Postgres versions (minor versions), added Postgres compatibility table to Parse Server docs (Corey Baker) [#7176](https://github.com/parse-community/parse-server/pull/7176)
 - Randomize test suite (Diamond Lewis) [#7265](https://github.com/parse-community/parse-server/pull/7265)
 - LDAP: Properly unbind client on group search error (Diamond Lewis) [#7265](https://github.com/parse-community/parse-server/pull/7265)
+- Improve data consistency in Push and Job Status update (Diamond Lewis) [#7267](https://github.com/parse-community/parse-server/pull/7267)
 ___
 ## 4.5.0
 [Full Changelog](https://github.com/parse-community/parse-server/compare/4.4.0...4.5.0)

src/Adapters/Cache/RedisCacheAdapter.js

@@ -1,6 +1,6 @@
 import redis from 'redis';
-import logger from '../../../logger';
-import { KeyPromiseQueue } from './KeyPromiseQueue';
+import logger from '../../logger';
+import { KeyPromiseQueue } from '../../KeyPromiseQueue';
 
 const DEFAULT_REDIS_TTL = 30 * 1000; // 30 seconds in milliseconds
 const FLUSH_DB_KEY = '__flush_db__';

src/StatusHandler.js

@@ -1,4 +1,5 @@
 import { md5Hash, newObjectId } from './cryptoUtils';
+import { KeyPromiseQueue } from './KeyPromiseQueue';
 import { logger } from './logger';
 import rest from './rest';
 import Auth from './Auth';
@@ -6,6 +7,9 @@ import Auth from './Auth';
 const PUSH_STATUS_COLLECTION = '_PushStatus';
 const JOB_STATUS_COLLECTION = '_JobStatus';
 
+const pushPromiseQueue = new KeyPromiseQueue();
+const jobPromiseQueue = new KeyPromiseQueue();
+
 const incrementOp = function (object = {}, key, amount = 1) {
   if (!object[key]) {
     object[key] = { __op: 'Increment', amount: amount };
@@ -28,22 +32,14 @@ export function flatten(array) {
 }
 
 function statusHandler(className, database) {
-  let lastPromise = Promise.resolve();
-
   function create(object) {
-    lastPromise = lastPromise.then(() => {
-      return database.create(className, object).then(() => {
-        return Promise.resolve(object);
-      });
+    return database.create(className, object).then(() => {
+      return Promise.resolve(object);
     });
-    return lastPromise;
   }
 
   function update(where, object) {
-    lastPromise = lastPromise.then(() => {
-      return database.update(className, where, object);
-    });
-    return lastPromise;
+    return jobPromiseQueue.enqueue(where.objectId, () => database.update(className, where, object));
   }
 
   return Object.freeze({
@@ -53,29 +49,21 @@ function statusHandler(className, database) {
 }
 
 function restStatusHandler(className, config) {
-  let lastPromise = Promise.resolve();
   const auth = Auth.master(config);
   function create(object) {
-    lastPromise = lastPromise.then(() => {
-      return rest.create(config, auth, className, object).then(({ response }) => {
-        // merge the objects
-        return Promise.resolve(Object.assign({}, object, response));
-      });
+    return rest.create(config, auth, className, object).then(({ response }) => {
+      return { ...object, ...response };
     });
-    return lastPromise;
   }
 
   function update(where, object) {
-    // TODO: when we have updateWhere, use that for proper interfacing
-    lastPromise = lastPromise.then(() => {
-      return rest
+    return pushPromiseQueue.enqueue(where.objectId, () =>
+      rest
         .update(config, auth, className, { objectId: where.objectId }, object)
         .then(({ response }) => {
-          // merge the objects
-          return Promise.resolve(Object.assign({}, object, response));
-        });
-    });
-    return lastPromise;
+          return { ...object, ...response };
+        })
+    );
   }
 
   return Object.freeze({