Fix session corruption due to database races

Prevent races between encrypt and decrypt calls, and other read/write
operations on the session store by serializing session io ops per
device.

Possible fix for #535

// FREEBIE

js/libtextsecure.js

@@ -35242,16 +35242,39 @@ axolotlInternal.RecipientRecord = function() {
     textsecure.storage.axolotl = new AxolotlStore();
     var axolotlInstance = axolotl.protocol(textsecure.storage.axolotl);
 
+    /*
+     * jobQueue manages multiple queues indexed by device to serialize
+     * session io ops on the database.
+     */
+    var jobQueue = {};
+    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;
+    }
+
     window.textsecure = window.textsecure || {};
     window.textsecure.protocol_wrapper = {
         decryptWhisperMessage: function(fromAddress, blob) {
-            return axolotlInstance.decryptWhisperMessage(fromAddress, getString(blob));
+            return queueJobForNumber(fromAddress, function() {
+                return axolotlInstance.decryptWhisperMessage(fromAddress, getString(blob));
+            });
         },
         closeOpenSessionForDevice: function(encodedNumber) {
-            return axolotlInstance.closeOpenSessionForDevice(encodedNumber);
+            return queueJobForNumber(encodedNumber, function() {
+                return axolotlInstance.closeOpenSessionForDevice(encodedNumber);
+            });
         },
         encryptMessageFor: function(deviceObject, pushMessageContent) {
-            return axolotlInstance.encryptMessageFor(deviceObject, pushMessageContent);
+            return queueJobForNumber(deviceObject.encodedNumber, function() {
+                return axolotlInstance.encryptMessageFor(deviceObject, pushMessageContent);
+            });
         },
         startWorker: function() {
             axolotlInstance.startWorker('/js/libaxolotl-worker.js');
@@ -35263,10 +35286,14 @@ axolotlInternal.RecipientRecord = function() {
             return axolotlInstance.createIdentityKeyRecvSocket();
         },
         hasOpenSession: function(encodedNumber) {
-            return axolotlInstance.hasOpenSession(encodedNumber);
+            return queueJobForNumber(encodedNumber, function() {
+                return axolotlInstance.hasOpenSession(encodedNumber);
+            });
         },
         getRegistrationId: function(encodedNumber) {
-            return axolotlInstance.getRegistrationId(encodedNumber);
+            return queueJobForNumber(encodedNumber, function() {
+                return axolotlInstance.getRegistrationId(encodedNumber);
+            });
         },
         handlePreKeyWhisperMessage: function(from, blob) {
             blob.mark();
@@ -35275,15 +35302,17 @@ axolotlInternal.RecipientRecord = function() {
                 // min version > 3 or max version < 3
                 throw new Error("Incompatible version byte");
             }
-            return axolotlInstance.handlePreKeyWhisperMessage(from, blob).catch(function(e) {
-                if (e.message === 'Unknown identity key') {
-                    blob.reset(); // restore the version byte.
-
-                    // create an error that the UI will pick up and ask the
-                    // user if they want to re-negotiate
-                    throw new textsecure.IncomingIdentityKeyError(from, blob.toArrayBuffer(), e.identityKey);
-                }
-                throw e;
+            return queueJobForNumber(from, function() {
+                return axolotlInstance.handlePreKeyWhisperMessage(from, blob).catch(function(e) {
+                    if (e.message === 'Unknown identity key') {
+                        blob.reset(); // restore the version byte.
+
+                        // create an error that the UI will pick up and ask the
+                        // user if they want to re-negotiate
+                        throw new textsecure.IncomingIdentityKeyError(from, blob.toArrayBuffer(), e.identityKey);
+                    }
+                    throw e;
+                });
             });
         }
     };

libtextsecure/axolotl_wrapper.js

@@ -9,16 +9,39 @@
     textsecure.storage.axolotl = new AxolotlStore();
     var axolotlInstance = axolotl.protocol(textsecure.storage.axolotl);
 
+    /*
+     * jobQueue manages multiple queues indexed by device to serialize
+     * session io ops on the database.
+     */
+    var jobQueue = {};
+    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;
+    }
+
     window.textsecure = window.textsecure || {};
     window.textsecure.protocol_wrapper = {
         decryptWhisperMessage: function(fromAddress, blob) {
-            return axolotlInstance.decryptWhisperMessage(fromAddress, getString(blob));
+            return queueJobForNumber(fromAddress, function() {
+                return axolotlInstance.decryptWhisperMessage(fromAddress, getString(blob));
+            });
         },
         closeOpenSessionForDevice: function(encodedNumber) {
-            return axolotlInstance.closeOpenSessionForDevice(encodedNumber);
+            return queueJobForNumber(encodedNumber, function() {
+                return axolotlInstance.closeOpenSessionForDevice(encodedNumber);
+            });
         },
         encryptMessageFor: function(deviceObject, pushMessageContent) {
-            return axolotlInstance.encryptMessageFor(deviceObject, pushMessageContent);
+            return queueJobForNumber(deviceObject.encodedNumber, function() {
+                return axolotlInstance.encryptMessageFor(deviceObject, pushMessageContent);
+            });
         },
         startWorker: function() {
             axolotlInstance.startWorker('/js/libaxolotl-worker.js');
@@ -30,10 +53,14 @@
             return axolotlInstance.createIdentityKeyRecvSocket();
         },
         hasOpenSession: function(encodedNumber) {
-            return axolotlInstance.hasOpenSession(encodedNumber);
+            return queueJobForNumber(encodedNumber, function() {
+                return axolotlInstance.hasOpenSession(encodedNumber);
+            });
         },
         getRegistrationId: function(encodedNumber) {
-            return axolotlInstance.getRegistrationId(encodedNumber);
+            return queueJobForNumber(encodedNumber, function() {
+                return axolotlInstance.getRegistrationId(encodedNumber);
+            });
         },
         handlePreKeyWhisperMessage: function(from, blob) {
             blob.mark();
@@ -42,15 +69,17 @@
                 // min version > 3 or max version < 3
                 throw new Error("Incompatible version byte");
             }
-            return axolotlInstance.handlePreKeyWhisperMessage(from, blob).catch(function(e) {
-                if (e.message === 'Unknown identity key') {
-                    blob.reset(); // restore the version byte.
+            return queueJobForNumber(from, function() {
+                return axolotlInstance.handlePreKeyWhisperMessage(from, blob).catch(function(e) {
+                    if (e.message === 'Unknown identity key') {
+                        blob.reset(); // restore the version byte.
 
-                    // create an error that the UI will pick up and ask the
-                    // user if they want to re-negotiate
-                    throw new textsecure.IncomingIdentityKeyError(from, blob.toArrayBuffer(), e.identityKey);
-                }
-                throw e;
+                        // create an error that the UI will pick up and ask the
+                        // user if they want to re-negotiate
+                        throw new textsecure.IncomingIdentityKeyError(from, blob.toArrayBuffer(), e.identityKey);
+                    }
+                    throw e;
+                });
             });
         }
     };