Improve queuing strategies in the app

js/models/conversations.js

@@ -736,23 +736,14 @@
     },
 
     queueJob(callback) {
-      const previous = this.pending || Promise.resolve();
+      this.jobQueue = this.jobQueue || new window.PQueue({ concurrency: 1 });
 
       const taskWithTimeout = textsecure.createTaskWithTimeout(
         callback,
         `conversation ${this.idForLogging()}`
       );
 
-      this.pending = previous.then(taskWithTimeout, taskWithTimeout);
-      const current = this.pending;
-
-      current.then(() => {
-        if (this.pending === current) {
-          delete this.pending;
-        }
-      });
-
-      return current;
+      return this.jobQueue.add(taskWithTimeout);
     },
 
     getRecipients() {

libtextsecure/account_manager.js

@@ -327,10 +327,11 @@
       });
     },
     queueTask(task) {
+      this.pendingQueue =
+        this.pendingQueue || new window.PQueue({ concurrency: 1 });
       const taskWithTimeout = textsecure.createTaskWithTimeout(task);
-      this.pending = this.pending.then(taskWithTimeout, taskWithTimeout);
 
-      return this.pending;
+      return this.pendingQueue.add(taskWithTimeout);
     },
     cleanSignedPreKeys() {
       const MINIMUM_KEYS = 3;

libtextsecure/libsignal-protocol.js

@@ -36486,14 +36486,10 @@ Internal.SessionLock = {};
 var jobQueue = {};
 
 Internal.SessionLock.queueJobForNumber = function queueJobForNumber(number, runJob) {
-     var runPrevious = jobQueue[number] || Promise.resolve();
-     var runCurrent = jobQueue[number] = runPrevious.then(runJob, runJob);
-     runCurrent.then(function() {
-         if (jobQueue[number] === runCurrent) {
-             delete jobQueue[number];
-         }
-     });
-     return runCurrent;
+     jobQueue[number] = jobQueue[number] || new window.PQueue({ concurrency: 1 });
+     var queue = jobQueue[number];
+
+     return queue.add(runJob);
 };
 
 })();

libtextsecure/message_receiver.js

@@ -134,10 +134,12 @@ function MessageReceiver(username, password, signalingKey, options = {}) {
   this.number = address.getName();
   this.deviceId = address.getDeviceId();
 
-  this.pending = Promise.resolve();
+  this.pendingQueue = new window.PQueue({ concurrency: 1 });
+  this.incomingQueue = new window.PQueue({ concurrency: 1 });
+  this.appQueue = new window.PQueue({ concurrency: 1 });
 
   if (options.retryCached) {
-    this.pending = this.queueAllCached();
+    this.pendingQueue.add(() => this.queueAllCached());
   }
 }
 
@@ -187,10 +189,6 @@ MessageReceiver.prototype.extend({
     // Because sometimes the socket doesn't properly emit its close event
     this._onClose = this.onclose.bind(this);
     this.wsr.addEventListener('close', this._onClose);
-
-    // Ensures that an immediate 'empty' event from the websocket will fire only after
-    //   all cached envelopes are processed.
-    this.incoming = [this.pending];
   },
   stopProcessing() {
     window.log.info('MessageReceiver: stopProcessing requested');
@@ -229,11 +227,7 @@ MessageReceiver.prototype.extend({
     window.log.error('websocket error');
   },
   dispatchAndWait(event) {
-    const promise = this.appPromise || Promise.resolve();
-    const appJobPromise = Promise.all(this.dispatchEvent(event));
-    const job = () => appJobPromise;
-
-    this.appPromise = promise.then(job, job);
+    this.appQueue.add(() => Promise.all(this.dispatchEvent(event)));
 
     return Promise.resolve();
   },
@@ -268,9 +262,6 @@ MessageReceiver.prototype.extend({
       });
   },
   handleRequest(request) {
-    this.incoming = this.incoming || [];
-    const lastPromise = _.last(this.incoming);
-
     // We do the message decryption here, instead of in the ordered pending queue,
     // to avoid exposing the time it took us to process messages through the time-to-ack.
 
@@ -279,132 +270,113 @@ MessageReceiver.prototype.extend({
       request.respond(200, 'OK');
 
       if (request.verb === 'PUT' && request.path === '/api/v1/queue/empty') {
-        this.onEmpty();
+        this.incomingQueue.add(() => this.onEmpty());
       }
       return;
     }
 
-    let promise;
-    const headers = request.headers || [];
-    if (headers.includes('X-Signal-Key: true')) {
-      promise = textsecure.crypto.decryptWebsocketMessage(
-        request.body,
-        this.signalingKey
-      );
-    } else {
-      promise = Promise.resolve(request.body.toArrayBuffer());
-    }
+    const job = async () => {
+      let plaintext;
+      const headers = request.headers || [];
 
-    promise = promise
-      .then(plaintext => {
+      if (headers.includes('X-Signal-Key: true')) {
+        plaintext = await textsecure.crypto.decryptWebsocketMessage(
+          request.body,
+          this.signalingKey
+        );
+      } else {
+        plaintext = request.body.toArrayBuffer();
+      }
+
+      try {
         const envelope = textsecure.protobuf.Envelope.decode(plaintext);
         // After this point, decoding errors are not the server's
         //   fault, and we should handle them gracefully and tell the
         //   user they received an invalid message
 
         if (this.isBlocked(envelope.source)) {
-          return request.respond(200, 'OK');
+          request.respond(200, 'OK');
+          return;
         }
 
         envelope.id = envelope.serverGuid || window.getGuid();
         envelope.serverTimestamp = envelope.serverTimestamp
           ? envelope.serverTimestamp.toNumber()
           : null;
 
-        return this.addToCache(envelope, plaintext).then(
-          async () => {
-            request.respond(200, 'OK');
-
-            // To ensure that we queue in the same order we receive messages
-            await lastPromise;
-            this.queueEnvelope(envelope);
-          },
-          error => {
-            request.respond(500, 'Failed to cache message');
-            window.log.error(
-              'handleRequest error trying to add message to cache:',
-              error && error.stack ? error.stack : error
-            );
-          }
-        );
-      })
-      .catch(e => {
+        try {
+          await this.addToCache(envelope, plaintext);
+          request.respond(200, 'OK');
+          this.queueEnvelope(envelope);
+        } catch (error) {
+          request.respond(500, 'Failed to cache message');
+          window.log.error(
+            'handleRequest error trying to add message to cache:',
+            error && error.stack ? error.stack : error
+          );
+        }
+      } catch (e) {
         request.respond(500, 'Bad encrypted websocket message');
         window.log.error(
           'Error handling incoming message:',
           e && e.stack ? e.stack : e
         );
         const ev = new Event('error');
         ev.error = e;
-        return this.dispatchAndWait(ev);
-      });
+        await this.dispatchAndWait(ev);
+      }
+    };
 
-    this.incoming.push(promise);
+    this.incomingQueue.add(job);
   },
   addToQueue(task) {
     this.count += 1;
-    this.pending = this.pending.then(task, task);
 
-    const { count, pending } = this;
+    const promise = this.pendingQueue.add(task);
 
-    const cleanup = () => {
+    const { count } = this;
+
+    const update = () => {
       this.updateProgress(count);
-      // We want to clear out the promise chain whenever possible because it could
-      //   lead to large memory usage over time:
-      //   https://github.com/nodejs/node/issues/6673#issuecomment-244331609
-      if (this.pending === pending) {
-        this.pending = Promise.resolve();
-      }
     };
 
-    pending.then(cleanup, cleanup);
+    promise.then(update, update);
 
-    return pending;
+    return promise;
   },
   onEmpty() {
-    const { incoming } = this;
-    this.incoming = [];
-
     const emitEmpty = () => {
       window.log.info("MessageReceiver: emitting 'empty' event");
       const ev = new Event('empty');
       this.dispatchAndWait(ev);
     };
 
-    const waitForApplication = async () => {
+    const waitForPendingQueue = () => {
       window.log.info(
         "MessageReceiver: finished processing messages after 'empty', now waiting for application"
       );
-      const promise = this.appPromise || Promise.resolve();
-      this.appPromise = Promise.resolve();
 
-      // We don't await here because we don't this to gate future message processing
-      promise.then(emitEmpty, emitEmpty);
+      // We don't await here because we don't want this to gate future message processing
+      this.appQueue.add(emitEmpty);
     };
 
-    const waitForEmptyQueue = () => {
-      // resetting count to zero so everything queued after this starts over again
-      this.count = 0;
+    const waitForIncomingQueue = () => {
+      this.addToQueue(waitForPendingQueue);
 
-      this.addToQueue(waitForApplication);
+      // Note: this.count is used in addToQueue
+      // Resetting count so everything from the websocket after this starts at zero
+      this.count = 0;
     };
 
-    // We first wait for all recently-received messages (this.incoming) to be queued,
-    //   then we queue a task to wait for the application to finish its processing, then
-    //   finally we emit the 'empty' event to the queue.
-    Promise.all(incoming).then(waitForEmptyQueue, waitForEmptyQueue);
+    this.incomingQueue.add(waitForIncomingQueue);
   },
   drain() {
-    const { incoming } = this;
-    this.incoming = [];
-
-    const queueDispatch = () =>
+    const waitForIncomingQueue = () =>
       this.addToQueue(() => {
         window.log.info('drained');
       });
 
-    // This promise will resolve when there are no more messages to be processed.
-    return Promise.all(incoming).then(queueDispatch, queueDispatch);
+    return this.incomingQueue.add(waitForIncomingQueue);
   },
   updateProgress(count) {
     // count by 10s

libtextsecure/sendmessage.js

@@ -249,23 +249,17 @@ MessageSender.prototype = {
   },
 
   queueJobForNumber(number, runJob) {
+    this.pendingMessages[number] =
+      this.pendingMessages[number] || new window.PQueue({ concurrency: 1 });
+
+    const queue = this.pendingMessages[number];
+
     const taskWithTimeout = textsecure.createTaskWithTimeout(
       runJob,
       `queueJobForNumber ${number}`
     );
 
-    const runPrevious = this.pendingMessages[number] || Promise.resolve();
-    this.pendingMessages[number] = runPrevious.then(
-      taskWithTimeout,
-      taskWithTimeout
-    );
-
-    const runCurrent = this.pendingMessages[number];
-    runCurrent.then(() => {
-      if (this.pendingMessages[number] === runCurrent) {
-        delete this.pendingMessages[number];
-      }
-    });
+    queue.add(taskWithTimeout);
   },
 
   uploadAttachments(message) {

ts/util/lint/exceptions.json

@@ -1336,31 +1336,31 @@
     "rule": "jQuery-wrap(",
     "path": "libtextsecure/message_receiver.js",
     "line": "  Promise.resolve(dcodeIO.ByteBuffer.wrap(string, 'binary').toArrayBuffer());",
-    "lineNumber": 145,
+    "lineNumber": 147,
     "reasonCategory": "falseMatch",
     "updated": "2018-09-19T18:13:29.628Z"
   },
   {
     "rule": "jQuery-wrap(",
     "path": "libtextsecure/message_receiver.js",
     "line": "  Promise.resolve(dcodeIO.ByteBuffer.wrap(arrayBuffer).toString('binary'));",
-    "lineNumber": 147,
+    "lineNumber": 149,
     "reasonCategory": "falseMatch",
     "updated": "2018-09-19T18:13:29.628Z"
   },
   {
     "rule": "jQuery-wrap(",
     "path": "libtextsecure/message_receiver.js",
     "line": "          const buffer = dcodeIO.ByteBuffer.wrap(ciphertext);",
-    "lineNumber": 833,
+    "lineNumber": 805,
     "reasonCategory": "falseMatch",
     "updated": "2018-09-19T18:13:29.628Z"
   },
   {
     "rule": "jQuery-wrap(",
     "path": "libtextsecure/message_receiver.js",
     "line": "        const buffer = dcodeIO.ByteBuffer.wrap(ciphertext);",
-    "lineNumber": 858,
+    "lineNumber": 830,
     "reasonCategory": "falseMatch",
     "updated": "2018-09-19T18:13:29.628Z"
   },