Bug 1207743 - Track recent push messages to avoid duplicate notifications on reconnect. r=dragana

MozReview-Commit-ID: KezPfa0yyO1

--HG--
extra : rebase_source : 86c3b0d78f2772369ebedd11d640a8b05e46d24f
extra : histedit_source : 531a495b7b3b3d8d4bf8b399f6beebf0805ac3bc

dom/push/PushRecord.jsm

@@ -43,6 +43,7 @@ function PushRecord(props) {
   this.authenticationSecret = props.authenticationSecret;
   this.systemRecord = !!props.systemRecord;
   this.appServerKey = props.appServerKey;
+  this.recentMessageIDs = props.recentMessageIDs;
   this.setQuota(props.quota);
   this.ctime = (typeof props.ctime === "number") ? props.ctime : 0;
 }
@@ -94,6 +95,29 @@ PushRecord.prototype = {
     this.lastPush = Date.now();
   },
 
+  /**
+   * Records a message ID sent to this push registration. We track the last few
+   * messages sent to each registration to avoid firing duplicate events for
+   * unacknowledged messages.
+   */
+  noteRecentMessageID(id) {
+    if (this.recentMessageIDs) {
+      this.recentMessageIDs.unshift(id);
+    } else {
+      this.recentMessageIDs = [id];
+    }
+    // Drop older message IDs from the end of the list.
+    let maxRecentMessageIDs = Math.min(
+      this.recentMessageIDs.length,
+      Math.max(prefs.get("maxRecentMessageIDsPerSubscription"), 0)
+    );
+    this.recentMessageIDs.length = maxRecentMessageIDs || 0;
+  },
+
+  hasRecentMessageID(id) {
+    return this.recentMessageIDs && this.recentMessageIDs.includes(id);
+  },
+
   reduceQuota() {
     if (!this.quotaApplies()) {
       return;

dom/push/PushService.jsm

@@ -860,7 +860,10 @@ this.PushService = {
         });
       });
     }).then(record => {
-      gPushNotifier.notifySubscriptionModified(record.scope, record.principal);
+      if (record) {
+        gPushNotifier.notifySubscriptionModified(record.scope,
+                                                 record.principal);
+      }
       return record;
     });
   },

dom/push/PushServiceWebSocket.jsm

@@ -715,13 +715,26 @@ this.PushServiceWebSocket = {
         update);
       return;
     }
+    function updateRecord(record) {
+      // Ignore messages that we've already processed. This can happen if the
+      // connection drops between notifying the service worker and acking the
+      // the message. In that case, the server will re-send the message on
+      // reconnect.
+      if (record.hasRecentMessageID(update.version)) {
+        console.warn("handleDataUpdate: Ignoring duplicate message",
+          update.version);
+        return null;
+      }
+      record.noteRecentMessageID(update.version);
+      return record;
+    }
     if (typeof update.data != "string") {
       promise = this._mainPushService.receivedPushMessage(
         update.channelID,
         update.version,
         null,
         null,
-        record => record
+        updateRecord
       );
     } else {
       let params = getCryptoParams(update.headers);
@@ -735,7 +748,7 @@ this.PushServiceWebSocket = {
           update.version,
           message,
           params,
-          record => record
+          updateRecord
         );
       } else {
         promise = Promise.reject(new Error("Invalid crypto headers"));

dom/push/test/test_utils.js

@@ -170,6 +170,7 @@ function setupPrefs() {
   return SpecialPowers.pushPrefEnv({"set": [
     ["dom.push.enabled", true],
     ["dom.push.connection.enabled", true],
+    ["dom.push.maxRecentMessageIDsPerSubscription", 0],
     ["dom.serviceWorkers.exemptFromPerDomainMax", true],
     ["dom.serviceWorkers.enabled", true],
     ["dom.serviceWorkers.testing.enabled", true]

dom/push/test/xpcshell/test_notification_duplicate.js

@@ -10,6 +10,7 @@ const userAgentID = '1500e7d9-8cbe-4ee6-98da-7fa5d6a39852';
 function run_test() {
   do_get_profile();
   setPrefs({
+    maxRecentMessageIDsPerSubscription: 4,
     userAgentID: userAgentID,
   });
   run_next_test();
@@ -20,19 +21,26 @@ add_task(function* test_notification_duplicate() {
   let db = PushServiceWebSocket.newPushDB();
   do_register_cleanup(() => {return db.drop().then(_ => db.close());});
   let records = [{
-    channelID: '8d2d9400-3597-4c5a-8a38-c546b0043bcc',
+    channelID: 'has-recents',
     pushEndpoint: 'https://example.org/update/1',
     scope: 'https://example.com/1',
     originAttributes: "",
-    version: 2,
+    recentMessageIDs: ['dupe'],
     quota: Infinity,
     systemRecord: true,
   }, {
-    channelID: '27d1e393-03ef-4c72-a5e6-9e890dfccad0',
+    channelID: 'no-recents',
     pushEndpoint: 'https://example.org/update/2',
     scope: 'https://example.com/2',
     originAttributes: "",
-    version: 2,
+    quota: Infinity,
+    systemRecord: true,
+  }, {
+    channelID: 'dropped-recents',
+    pushEndpoint: 'https://example.org/update/3',
+    scope: 'https://example.com/3',
+    originAttributes: '',
+    recentMessageIDs: ['newest', 'newer', 'older', 'oldest'],
     quota: Infinity,
     systemRecord: true,
   }];
@@ -40,11 +48,48 @@ add_task(function* test_notification_duplicate() {
     yield db.put(record);
   }
 
-  let notifyPromise = promiseObserverNotification(PushServiceComponent.pushTopic);
+  let testData = [{
+    channelID: 'has-recents',
+    updates: 1,
+    acks: [{
+      version: 'dupe',
+      code: 102,
+    }, {
+      version: 'not-dupe',
+      code: 100,
+    }],
+    recents: ['not-dupe', 'dupe'],
+  }, {
+    channelID: 'no-recents',
+    updates: 1,
+    acks: [{
+      version: 'not-dupe',
+      code: 100,
+    }],
+    recents: ['not-dupe'],
+  }, {
+    channelID: 'dropped-recents',
+    acks: [{
+      version: 'overflow',
+      code: 100,
+    }, {
+      version: 'oldest',
+      code: 100,
+    }],
+    updates: 2,
+    recents: ['oldest', 'overflow', 'newest', 'newer'],
+  }];
 
-  let acks = 0;
+  let expectedUpdates = testData.reduce((sum, {updates}) => sum + updates, 0);
+  let notifiedScopes = [];
+  let notifyPromise = promiseObserverNotification(PushServiceComponent.pushTopic, (subject, data) => {
+    notifiedScopes.push(data);
+    return notifiedScopes.length == expectedUpdates;
+  });
+
+  let expectedAcks = testData.reduce((sum, {acks}) => sum + acks.length, 0);
   let ackDone;
-  let ackPromise = new Promise(resolve => ackDone = after(2, resolve));
+  let ackPromise = new Promise(resolve => ackDone = after(expectedAcks, resolve));
   PushService.init({
     serverURI: "wss://push.example.org/",
     db,
@@ -55,19 +100,31 @@ add_task(function* test_notification_duplicate() {
             messageType: 'hello',
             status: 200,
             uaid: userAgentID,
+            use_webpush: true,
           }));
-          this.serverSendMsg(JSON.stringify({
-            messageType: 'notification',
-            updates: [{
-              channelID: '8d2d9400-3597-4c5a-8a38-c546b0043bcc',
-              version: 2
-            }, {
-              channelID: '27d1e393-03ef-4c72-a5e6-9e890dfccad0',
-              version: 3
-            }]
-          }));
+          for (let {channelID, acks} of testData) {
+            for (let {version} of acks) {
+              this.serverSendMsg(JSON.stringify({
+                messageType: 'notification',
+                channelID: channelID,
+                version: version,
+              }))
+            }
+          }
+        },
+        onACK(request) {
+          let [ack] = request.updates;
+          let expectedData = testData.find(test =>
+            test.channelID == ack.channelID);
+          ok(expectedData, `Unexpected channel ${ack.channelID}`);
+          let expectedAck = expectedData.acks.find(expectedAck =>
+            expectedAck.version == ack.version);
+          ok(expectedAck, `Unexpected ack for message ${
+            ack.version} on ${ack.channelID}`);
+          equal(expectedAck.code, ack.code, `Wrong ack status for message ${
+            ack.version} on ${ack.channelID}`);
+          ackDone();
         },
-        onACK: ackDone
       });
     }
   });
@@ -75,11 +132,9 @@ add_task(function* test_notification_duplicate() {
   yield notifyPromise;
   yield ackPromise;
 
-  let staleRecord = yield db.getByKeyID(
-    '8d2d9400-3597-4c5a-8a38-c546b0043bcc');
-  strictEqual(staleRecord.version, 2, 'Wrong stale record version');
-
-  let updatedRecord = yield db.getByKeyID(
-    '27d1e393-03ef-4c72-a5e6-9e890dfccad0');
-  strictEqual(updatedRecord.version, 3, 'Wrong updated record version');
+  for (let {channelID, recents} of testData) {
+    let record = yield db.getByKeyID(channelID);
+    deepEqual(record.recentMessageIDs, recents,
+      `Wrong recent message IDs for ${channelID}`);
+  }
 });

mobile/android/app/mobile.js

@@ -902,6 +902,7 @@ pref("dom.push.debug", false);
 // The upstream autopush endpoint must have the Google API key corresponding to
 // the App's sender ID; we bake this assumption directly into the URL.
 pref("dom.push.serverURL", "https://updates-autopush.stage.mozaws.net/v1/gcm/@MOZ_ANDROID_GCM_SENDERID@");
+pref("dom.push.maxRecentMessageIDsPerSubscription", 0);
 
 #ifdef MOZ_ANDROID_GCM
 pref("dom.push.enabled", true);

modules/libpref/init/all.js

@@ -4711,6 +4711,10 @@ pref("dom.push.userAgentID", "");
 // without user interaction.
 pref("dom.push.maxQuotaPerSubscription", 16);
 
+// The maximum number of recent message IDs to store for each push
+// subscription, to avoid duplicates for unacknowledged messages.
+pref("dom.push.maxRecentMessageIDsPerSubscription", 10);
+
 // The delay between receiving a push message and updating the quota for a
 // subscription.
 pref("dom.push.quotaUpdateDelay", 3000); // 3 seconds