Improvements for sending push performance (#4122)

* Adds test for stalled pushStatus when audience is empty

* fixup! Adds test for stalled pushStatus when audience is empty

* Do not enqueue when count is 0, enforce deviceToken exists, stop badge ordering

spec/PushController.spec.js

@@ -845,6 +845,72 @@ describe('PushController', () => {
       fail('should not fail');
       done();
     });
+  });
 
+  it('should mark the _PushStatus as succeeded when audience has no deviceToken', (done) => {
+    var auth = {
+      isMaster: true
+    }
+    var pushAdapter = {
+      send: function(body, installations) {
+        const promises = installations.map((device) => {
+          if (!device.deviceToken) {
+            // Simulate error when device token is not set
+            return Promise.reject();
+          }
+          return Promise.resolve({
+            transmitted: true,
+            device: device,
+          })
+        });
+
+        return Promise.all(promises);
+      },
+      getValidPushTypes: function() {
+        return ["ios"];
+      }
+    }
+
+    var pushController = new PushController();
+    const payload = {
+      data: {
+        alert: 'hello',
+      },
+      push_time: new Date().getTime() / 1000
+    }
+
+    var installations = [];
+    while(installations.length != 5) {
+      const installation = new Parse.Object("_Installation");
+      installation.set("installationId", "installation_" + installations.length);
+      installation.set("badge", installations.length);
+      installation.set("originalBadge", installations.length);
+      installation.set("deviceType", "ios");
+      installations.push(installation);
+    }
+
+    reconfigureServer({
+      push: { adapter: pushAdapter }
+    }).then(() => {
+      var config = new Config(Parse.applicationId);
+      return Parse.Object.saveAll(installations).then(() => {
+        return pushController.sendPush(payload, {}, config, auth)
+          .then(() => { done.fail('should not success') })
+          .catch(() => {})
+      }).then(() => new Promise(resolve => setTimeout(resolve, 100)));
+    }).then(() => {
+      const query = new Parse.Query('_PushStatus');
+      return query.find({useMasterKey: true}).then((results) => {
+        expect(results.length).toBe(1);
+        const pushStatus = results[0];
+        expect(pushStatus.get('numSent')).toBe(0);
+        expect(pushStatus.get('status')).toBe('failed');
+        done();
+      });
+    }).catch((err) => {
+      console.error(err);
+      fail('should not fail');
+      done();
+    });
   });
 });

src/Controllers/PushController.js

@@ -1,9 +1,9 @@
 import { Parse }              from 'parse/node';
-import deepcopy               from 'deepcopy';
 import RestQuery              from '../RestQuery';
 import RestWrite              from '../RestWrite';
 import { master }             from '../Auth';
 import { pushStatusHandler }  from '../StatusHandler';
+import { applyDeviceTokenExists } from '../Push/utils';
 
 export class PushController {
 
@@ -23,6 +23,7 @@ export class PushController {
     let badgeUpdate = () => {
       return Promise.resolve();
     }
+
     if (body.data && body.data.badge) {
       const badge = body.data.badge;
       let restUpdate = {};
@@ -33,8 +34,9 @@ export class PushController {
       } else {
         throw "Invalid value for badge, expected number or 'Increment'";
       }
-      const updateWhere = deepcopy(where);
 
+      // Force filtering on only valid device tokens
+      const updateWhere = applyDeviceTokenExists(where);
       badgeUpdate = () => {
         // Build a real RestQuery so we can use it in RestWrite
         const restQuery = new RestQuery(config, master(config), '_Installation', updateWhere);

src/Push/PushQueue.js

@@ -1,7 +1,6 @@
-import { ParseMessageQueue }  from '../ParseMessageQueue';
+import { ParseMessageQueue }      from '../ParseMessageQueue';
-import rest                   from '../rest';
+import rest                       from '../rest';
-import { isPushIncrementing } from './utils';
+import { applyDeviceTokenExists } from './utils';
-import deepcopy               from 'deepcopy';
 
 const PUSH_CHANNEL = 'parse-server-push';
 const DEFAULT_BATCH_SIZE = 100;
@@ -25,13 +24,11 @@ export class PushQueue {
 
   enqueue(body, where, config, auth, pushStatus) {
     const limit = this.batchSize;
-    // Order by badge (because the payload is badge dependant)
+
-    // and createdAt to fix the order
+    where = applyDeviceTokenExists(where);
-    const order = isPushIncrementing(body) ? 'badge,createdAt' : 'createdAt';
+
-    where = deepcopy(where);
+    // Order by objectId so no impact on the DB
-    if (!where.hasOwnProperty('deviceToken')) {
+    const order = 'objectId';
-      where['deviceToken'] = {'$exists': true};
-    }
     return Promise.resolve().then(() => {
       return rest.find(config,
         auth,
@@ -39,7 +36,7 @@ export class PushQueue {
         where,
         {limit: 0, count: true});
     }).then(({results, count}) => {
-      if (!results) {
+      if (!results || count == 0) {
         return Promise.reject({error: 'PushController: no results in query'})
       }
       pushStatus.setRunning(count);

src/Push/PushWorker.js

@@ -6,7 +6,7 @@ import Config                 from '../Config';
 import { PushAdapter }        from '../Adapters/Push/PushAdapter';
 import rest                   from '../rest';
 import { pushStatusHandler }  from '../StatusHandler';
-import { isPushIncrementing } from './utils';
+import * as utils             from './utils';
 import { ParseMessageQueue }  from '../ParseMessageQueue';
 import { PushQueue }          from './PushQueue';
 
@@ -51,7 +51,7 @@ export class PushWorker {
   run({ body, query, pushStatus, applicationId }: any): Promise<*> {
     const config = new Config(applicationId);
     const auth = master(config);
-    const where  = query.where;
+    const where = utils.applyDeviceTokenExists(query.where);
     delete query.where;
     return rest.find(config, auth, '_Installation', where, query).then(({results}) => {
       if (results.length == 0) {
@@ -65,7 +65,7 @@ export class PushWorker {
 
   sendToAdapter(body: any, installations: any[], pushStatus: any, config: Config): Promise<*> {
     pushStatus = pushStatusHandler(config, pushStatus.objectId);
-    if (!isPushIncrementing(body)) {
+    if (!utils.isPushIncrementing(body)) {
       return this.adapter.send(body, installations, pushStatus.objectId).then((results) => {
         return pushStatus.trackSent(results);
       });

src/Push/utils.js

@@ -1,4 +1,5 @@
-import Parse from 'parse/node';
+import Parse    from 'parse/node';
+import deepcopy from 'deepcopy';
 
 export function isPushIncrementing(body) {
   return body.data &&
@@ -28,3 +29,11 @@ export function validatePushType(where = {}, validPushTypes = []) {
     }
   }
 }
+
+export function applyDeviceTokenExists(where) {
+  where = deepcopy(where);
+  if (!where.hasOwnProperty('deviceToken')) {
+    where['deviceToken'] = {'$exists': true};
+  }
+  return where;
+}