Two fixes for messages causing errors

* Queue delivery receipt sends, only start after we get 'empty'
* Retry cached two minutes after empty, or any post-empty message

js/background.js

@@ -17,6 +17,10 @@
   'use strict';
 
   const eventHandlerQueue = new window.PQueue({ concurrency: 1 });
+  const deliveryReceiptQueue = new window.PQueue({
+    concurrency: 1,
+  });
+  deliveryReceiptQueue.pause();
 
   // Globally disable drag and drop
   document.body.addEventListener(
@@ -824,6 +828,7 @@
       serverTrustRoot: window.getServerTrustRoot(),
     };
 
+    deliveryReceiptQueue.pause(); // avoid flood of delivery receipts until we catch up
     Whisper.Notifications.disable(); // avoid notification flood until empty
 
     // initialize the socket and start listening for messages
@@ -996,6 +1001,7 @@
       }
     }, 500);
 
+    deliveryReceiptQueue.start();
     Whisper.Notifications.enable();
   }
   function onReconnect() {
@@ -1003,6 +1009,7 @@
     //   scenarios where we're coming back from sleep, we can get offline/online events
     //   very fast, and it looks like a network blip. But we need to suppress
     //   notifications in these scenarios too. So we listen for 'reconnect' events.
+    deliveryReceiptQueue.pause();
     Whisper.Notifications.disable();
   }
   function onProgress(ev) {
@@ -1545,6 +1552,7 @@
       return message;
     }
 
+    deliveryReceiptQueue.add(async () => {
       try {
         const { wrap, sendOptions } = ConversationController.prepareForSend(
           data.source
@@ -1564,6 +1572,7 @@
           error && error.stack ? error.stack : error
         );
       }
+    });
 
     return message;
   }

libtextsecure/message_receiver.js

@@ -14,6 +14,7 @@
 /* eslint-disable more/no-then */
 
 const WORKER_TIMEOUT = 60 * 1000; // one minute
+const RETRY_TIMEOUT = 2 * 60 * 1000;
 
 const _utilWorker = new Worker('js/util_worker.js');
 const _jobs = Object.create(null);
@@ -167,6 +168,7 @@ MessageReceiver.prototype.extend({
       this.dispatchEvent(ev);
     }
 
+    this.isEmptied = false;
     this.hasConnected = true;
 
     if (this.socket && this.socket.readyState !== WebSocket.CLOSED) {
@@ -218,6 +220,8 @@ MessageReceiver.prototype.extend({
       this.wsr.close(3000, 'called close');
     }
 
+    this.clearRetryTimeout();
+
     return this.drain();
   },
   onopen() {
@@ -308,6 +312,9 @@ MessageReceiver.prototype.extend({
           await this.addToCache(envelope, plaintext);
           request.respond(200, 'OK');
           this.queueEnvelope(envelope);
+
+          this.clearRetryTimeout();
+          this.maybeScheduleRetryTimeout();
         } catch (error) {
           request.respond(500, 'Failed to cache message');
           window.log.error(
@@ -349,6 +356,8 @@ MessageReceiver.prototype.extend({
       window.log.info("MessageReceiver: emitting 'empty' event");
       const ev = new Event('empty');
       this.dispatchAndWait(ev);
+      this.isEmptied = true;
+      this.maybeScheduleRetryTimeout();
     };
 
     const waitForPendingQueue = () => {
@@ -465,6 +474,19 @@ MessageReceiver.prototype.extend({
 
     return envelope.id;
   },
+  clearRetryTimeout() {
+    if (this.retryCachedTimeout) {
+      clearInterval(this.retryCachedTimeout);
+      this.retryCachedTimeout = null;
+    }
+  },
+  maybeScheduleRetryTimeout() {
+    if (this.isEmptied) {
+      this.retryCachedTimeout = setTimeout(() => {
+        this.pendingQueue.add(() => this.queueAllCached());
+      }, RETRY_TIMEOUT);
+    }
+  },
   async getAllFromCache() {
     window.log.info('getAllFromCache');
     const count = await textsecure.storage.unprocessed.getCount();