Whenever adding something to a queue, include a timeout

No more wedged queues!

FREEBIE

Gruntfile.js

@@ -61,6 +61,7 @@ module.exports = function(grunt) {
           'libtextsecure/sync_request.js',
           'libtextsecure/contacts_parser.js',
           'libtextsecure/ProvisioningCipher.js',
+          'libtextsecure/task_with_timeout.js',
         ],
         dest: 'js/libtextsecure.js',
       },

js/libtextsecure.js

@@ -38094,7 +38094,8 @@ var TextSecureServer = (function() {
             }.bind(this));
         },
         queueTask: function(task) {
-            return this.pending = this.pending.then(task, task);
+            var taskWithTimeout = textsecure.createTaskWithTimeout(task);
+            return this.pending = this.pending.then(taskWithTimeout, taskWithTimeout);
         },
         cleanSignedPreKeys: function() {
             var nextSignedKeyId = textsecure.storage.get('signedKeyId');
@@ -38420,21 +38421,29 @@ MessageReceiver.prototype.extend({
         return textsecure.storage.unprocessed.remove(id);
     },
     queueDecryptedEnvelope: function(envelope, plaintext) {
-        console.log('queueing decrypted envelope', this.getEnvelopeId(envelope));
-        var handleDecryptedEnvelope = this.handleDecryptedEnvelope.bind(this, envelope, plaintext);
-        this.pending = this.pending.then(handleDecryptedEnvelope, handleDecryptedEnvelope);
+        var id = this.getEnvelopeId(envelope);
+        console.log('queueing decrypted envelope', id);
+
+        var task = this.handleDecryptedEnvelope.bind(this, envelope, plaintext);
+        var taskWithTimeout = textsecure.createTaskWithTimeout(task, 'queueEncryptedEnvelope ' + id);
+
+        this.pending = this.pending.then(taskWithTimeout, taskWithTimeout);
 
         return this.pending.catch(function(error) {
-            console.log('queueDecryptedEnvelope error:', error && error.stack ? error.stack : error);
+            console.log('queueDecryptedEnvelope error handling envelope', id, ':', error && error.stack ? error.stack : error);
         });
     },
     queueEnvelope: function(envelope) {
-        console.log('queueing envelope', this.getEnvelopeId(envelope));
-        var handleEnvelope = this.handleEnvelope.bind(this, envelope);
-        this.pending = this.pending.then(handleEnvelope, handleEnvelope);
+        var id = this.getEnvelopeId(envelope);
+        console.log('queueing envelope', id);
+
+        var task = this.handleEnvelope.bind(this, envelope);
+        var taskWithTimeout = textsecure.createTaskWithTimeout(task, 'queueEnvelope ' + id);
+
+        this.pending = this.pending.then(taskWithTimeout, taskWithTimeout);
 
         return this.pending.catch(function(error) {
-            console.log('queueEnvelope error:', error && error.stack ? error.stack : error);
+            console.log('queueEnvelope error handling envelope', id, ':', error && error.stack ? error.stack : error);
         });
     },
     // Same as handleEnvelope, just without the decryption step. Necessary for handling
@@ -39325,8 +39334,10 @@ MessageSender.prototype = {
     },
 
     queueJobForNumber: function(number, runJob) {
+        var taskWithTimeout = textsecure.createTaskWithTimeout(runJob, 'queueJobForNumber ' + number);
+
         var runPrevious = this.pendingMessages[number] || Promise.resolve();
-        var runCurrent = this.pendingMessages[number] = runPrevious.then(runJob, runJob);
+        var runCurrent = this.pendingMessages[number] = runPrevious.then(taskWithTimeout, taskWithTimeout);
         runCurrent.then(function() {
             if (this.pendingMessages[number] === runCurrent) {
                 delete this.pendingMessages[number];
@@ -39969,4 +39980,70 @@ libsignal.ProvisioningCipher = function() {
 };
 
 })();
+
+/*
+ * vim: ts=4:sw=4:expandtab
+ */
+(function () {
+    window.textsecure = window.textsecure || {};
+
+    window.textsecure.createTaskWithTimeout = function(task, id, options) {
+        options = options || {};
+        options.timeout = options.timeout || (1000 * 60 * 2); // two minutes
+
+        var errorForStack = new Error('for stack');
+        return function() {
+            return new Promise(function(resolve, reject) {
+                var complete = false;
+                var timer = setTimeout(function() {
+                    if (!complete) {
+                        var message =
+                            (id || '')
+                            + ' task did not complete in time. Calling stack: '
+                            + errorForStack.stack;
+
+                        console.log(message);
+                        return reject(new Error(message));
+                    }
+                }.bind(this), options.timeout);
+                var clearTimer = function() {
+                    try {
+                        var localTimer = timer;
+                        if (localTimer) {
+                            timer = null;
+                            clearTimeout(localTimer);
+                        }
+                    }
+                    catch (error) {
+                        console.log(
+                            id || '',
+                            'task ran into problem canceling timer. Calling stack:',
+                            errorForStack.stack
+                        );
+                    }
+                };
+
+                var success = function(result) {
+                    clearTimer();
+                    complete = true;
+                    return resolve(result);
+                };
+                var failure = function(error) {
+                    clearTimer();
+                    complete = true;
+                    return reject(error);
+                };
+
+                var promise = task();
+                if (!promise || !promise.then) {
+                    clearTimer();
+                    complete = true;
+                    return resolve(promise);
+                }
+
+                return promise.then(success, failure);
+            });
+        };
+    };
+})();
 })();

js/models/conversations.js

@@ -411,7 +411,10 @@
 
     queueJob: function(callback) {
         var previous = this.pending || Promise.resolve();
-        var current = this.pending = previous.then(callback, callback);
+
+        var taskWithTimeout = textsecure.createTaskWithTimeout(callback, 'conversation ' + this.id);
+
+        var current = this.pending = previous.then(taskWithTimeout, taskWithTimeout);
 
         current.then(function() {
             if (this.pending === current) {

js/models/messages.js

@@ -480,9 +480,6 @@
                         };
 
                         message.save().then(function() {
-
-                            // throw new Error('Something went wrong!');
-
                             conversation.save().then(function() {
                                 try {
                                     conversation.trigger('newmessage', message);

libtextsecure/account_manager.js

@@ -146,7 +146,8 @@
             }.bind(this));
         },
         queueTask: function(task) {
-            return this.pending = this.pending.then(task, task);
+            var taskWithTimeout = textsecure.createTaskWithTimeout(task);
+            return this.pending = this.pending.then(taskWithTimeout, taskWithTimeout);
         },
         cleanSignedPreKeys: function() {
             var nextSignedKeyId = textsecure.storage.get('signedKeyId');

libtextsecure/message_receiver.js

@@ -182,21 +182,29 @@ MessageReceiver.prototype.extend({
         return textsecure.storage.unprocessed.remove(id);
     },
     queueDecryptedEnvelope: function(envelope, plaintext) {
-        console.log('queueing decrypted envelope', this.getEnvelopeId(envelope));
-        var handleDecryptedEnvelope = this.handleDecryptedEnvelope.bind(this, envelope, plaintext);
-        this.pending = this.pending.then(handleDecryptedEnvelope, handleDecryptedEnvelope);
+        var id = this.getEnvelopeId(envelope);
+        console.log('queueing decrypted envelope', id);
+
+        var task = this.handleDecryptedEnvelope.bind(this, envelope, plaintext);
+        var taskWithTimeout = textsecure.createTaskWithTimeout(task, 'queueEncryptedEnvelope ' + id);
+
+        this.pending = this.pending.then(taskWithTimeout, taskWithTimeout);
 
         return this.pending.catch(function(error) {
-            console.log('queueDecryptedEnvelope error:', error && error.stack ? error.stack : error);
+            console.log('queueDecryptedEnvelope error handling envelope', id, ':', error && error.stack ? error.stack : error);
         });
     },
     queueEnvelope: function(envelope) {
-        console.log('queueing envelope', this.getEnvelopeId(envelope));
-        var handleEnvelope = this.handleEnvelope.bind(this, envelope);
-        this.pending = this.pending.then(handleEnvelope, handleEnvelope);
+        var id = this.getEnvelopeId(envelope);
+        console.log('queueing envelope', id);
+
+        var task = this.handleEnvelope.bind(this, envelope);
+        var taskWithTimeout = textsecure.createTaskWithTimeout(task, 'queueEnvelope ' + id);
+
+        this.pending = this.pending.then(taskWithTimeout, taskWithTimeout);
 
         return this.pending.catch(function(error) {
-            console.log('queueEnvelope error:', error && error.stack ? error.stack : error);
+            console.log('queueEnvelope error handling envelope', id, ':', error && error.stack ? error.stack : error);
         });
     },
     // Same as handleEnvelope, just without the decryption step. Necessary for handling

libtextsecure/sendmessage.js

@@ -149,8 +149,10 @@ MessageSender.prototype = {
     },
 
     queueJobForNumber: function(number, runJob) {
+        var taskWithTimeout = textsecure.createTaskWithTimeout(runJob, 'queueJobForNumber ' + number);
+
         var runPrevious = this.pendingMessages[number] || Promise.resolve();
-        var runCurrent = this.pendingMessages[number] = runPrevious.then(runJob, runJob);
+        var runCurrent = this.pendingMessages[number] = runPrevious.then(taskWithTimeout, taskWithTimeout);
         runCurrent.then(function() {
             if (this.pendingMessages[number] === runCurrent) {
                 delete this.pendingMessages[number];

libtextsecure/task_with_timeout.js

@@ -0,0 +1,65 @@
+/*
+ * vim: ts=4:sw=4:expandtab
+ */
+(function () {
+    window.textsecure = window.textsecure || {};
+
+    window.textsecure.createTaskWithTimeout = function(task, id, options) {
+        options = options || {};
+        options.timeout = options.timeout || (1000 * 60 * 2); // two minutes
+
+        var errorForStack = new Error('for stack');
+        return function() {
+            return new Promise(function(resolve, reject) {
+                var complete = false;
+                var timer = setTimeout(function() {
+                    if (!complete) {
+                        var message =
+                            (id || '')
+                            + ' task did not complete in time. Calling stack: '
+                            + errorForStack.stack;
+
+                        console.log(message);
+                        return reject(new Error(message));
+                    }
+                }.bind(this), options.timeout);
+                var clearTimer = function() {
+                    try {
+                        var localTimer = timer;
+                        if (localTimer) {
+                            timer = null;
+                            clearTimeout(localTimer);
+                        }
+                    }
+                    catch (error) {
+                        console.log(
+                            id || '',
+                            'task ran into problem canceling timer. Calling stack:',
+                            errorForStack.stack
+                        );
+                    }
+                };
+
+                var success = function(result) {
+                    clearTimer();
+                    complete = true;
+                    return resolve(result);
+                };
+                var failure = function(error) {
+                    clearTimer();
+                    complete = true;
+                    return reject(error);
+                };
+
+                var promise = task();
+                if (!promise || !promise.then) {
+                    clearTimer();
+                    complete = true;
+                    return resolve(promise);
+                }
+
+                return promise.then(success, failure);
+            });
+        };
+    };
+})();

libtextsecure/test/index.html

@@ -28,9 +28,10 @@
   <script type="text/javascript" src="../stringview.js" data-cover></script>
   <script type="text/javascript" src="../api.js"></script>
   <script type="text/javascript" src="../sendmessage.js" data-cover></script>
-  <script type="text/javascript" src="../event_target.js" data-></script>
-  <script type="text/javascript" src="../account_manager.js" data-></script>
-  <script type="text/javascript" src="../contacts_parser.js"></script>
+  <script type="text/javascript" src="../event_target.js" data-cover></script>
+  <script type="text/javascript" src="../account_manager.js" data-cover></script>
+  <script type="text/javascript" src="../contacts_parser.js" data-cover></script>
+  <script type="text/javascript" src="../task_with_timeout.js" data-cover></script>
 
   <script type="text/javascript" src="fake_api.js"></script>
   <script type="text/javascript" src="helpers_test.js"></script>
@@ -39,5 +40,6 @@
   <script type="text/javascript" src="contacts_parser_test.js"></script>
   <script type="text/javascript" src="generate_keys_test.js"></script>
   <script type="text/javascript" src="websocket-resources_test.js"></script>
+  <script type="text/javascript" src="task_with_timeout_test.js"></script>
 </body>
 </html>

libtextsecure/test/task_with_timeout_test.js

@@ -0,0 +1,60 @@
+'use strict';
+
+describe('createTaskWithTimeout', function() {
+    it('resolves when promise resolves', function() {
+        var task = function() {
+            return Promise.resolve('hi!');
+        };
+        var taskWithTimeout = textsecure.createTaskWithTimeout(task);
+
+        return taskWithTimeout().then(function(result) {
+            assert.strictEqual(result, 'hi!')
+        });
+    });
+    it('flows error from promise back', function() {
+        var error = new Error('original');
+        var task = function() {
+            return Promise.reject(error);
+        };
+        var taskWithTimeout = textsecure.createTaskWithTimeout(task);
+
+        return taskWithTimeout().catch(function(flowedError) {
+            assert.strictEqual(error, flowedError);
+        });
+    });
+    it('rejects if promise takes too long', function() {
+        var error = new Error('original');
+        var complete = false;
+        var task = function() {
+            return new Promise(function(resolve) {
+                setTimeout(function() {
+                    completed = true;
+                    resolve();
+                }, 3000);
+            });
+        };
+        var taskWithTimeout = textsecure.createTaskWithTimeout(task, this.name, {
+            timeout: 10
+        });
+
+        return taskWithTimeout().then(function() {
+            throw new Error('it was not supposed to resolve!');
+        }, function() {
+            assert.strictEqual(complete, false);
+        });
+    });
+    it('resolves if task returns something falsey', function() {
+        var task = function() {};
+        var taskWithTimeout = textsecure.createTaskWithTimeout(task);
+        return taskWithTimeout();
+    });
+    it('resolves if task returns a non-promise', function() {
+        var task = function() {
+            return 'hi!';
+        };
+        var taskWithTimeout = textsecure.createTaskWithTimeout(task);
+        return taskWithTimeout().then(function(result) {
+            assert.strictEqual(result, 'hi!')
+        });
+    });
+});