Avoid a packetstorm of device queries on startup

Two main changes here:
 * when we get an m.new_device event for a device we know about, ignore it
 * Batch up the m.new_device events received during initialsync and spam out
   all the queries at once.

lib/crypto/algorithms/base.js

@@ -88,15 +88,6 @@ EncryptionAlgorithm.prototype.onRoomMembership = function(
     event, member, oldMembership
 ) {};
 
-/**
- * Called when a new device announces itself in the room
- *
- * @param {string} userId    owner of the device
- * @param {string} deviceId  deviceId of the device
- */
-EncryptionAlgorithm.prototype.onNewDevice = function(userId, deviceId) {};
-
-
 /**
  * base type for decryption implementations
  *

lib/crypto/index.js

@@ -54,6 +54,10 @@ function Crypto(baseApis, eventEmitter, sessionStore, userId, deviceId) {
     this._userId = userId;
     this._deviceId = deviceId;
 
+    this._initialSyncCompleted = false;
+    // userId -> deviceId -> true
+    this._pendingNewDevices = {};
+
     this._olmDevice = new OlmDevice(sessionStore);
 
     // EncryptionAlgorithm instance for each room
@@ -272,24 +276,25 @@ Crypto.prototype.downloadKeys = function(userIds, forceDownload) {
 
     // map from userid -> deviceid -> DeviceInfo
     var stored = {};
+    function storeDev(userId, dev) {
+        stored[userId][dev.deviceId] = dev;
+    }
 
     // list of userids we need to download keys for
     var downloadUsers = [];
 
-    for (var i = 0; i < userIds.length; ++i) {
+    if (forceDownload) {
-        var userId = userIds[i];
+        downloadUsers = userIds;
-        stored[userId] = {};
+    } else {
+        for (var i = 0; i < userIds.length; ++i) {
+            var userId = userIds[i];
+            var devices = this.getStoredDevicesForUser(userId);
 
-        var devices = this.getStoredDevicesForUser(userId);
+            if (!devices) {
-
+                downloadUsers.push(userId);
-        if (!devices || forceDownload) {
+            } else {
-            downloadUsers.push(userId);
+                stored[userId] = {};
-        }
+                devices.map(storeDev.bind(null, userId));
-
-        if (devices) {
-            for (var j = 0; j < devices.length; ++j) {
-                var dev = devices[j];
-                stored[userId][dev.deviceId] = dev;
             }
         }
     }
@@ -298,30 +303,79 @@ Crypto.prototype.downloadKeys = function(userIds, forceDownload) {
         return q(stored);
     }
 
-    return this._baseApis.downloadKeysForUsers(
+    var r = this._doKeyDownloadForUsers(downloadUsers);
+    var promises = [];
+    downloadUsers.map(function(u) {
+        promises.push(r[u].catch(function(e) {
+            console.warn('Error downloading keys for user ' + u + ':', e);
+        }).then(function() {
+            stored[u] = {};
+            var devices = self.getStoredDevicesForUser(u) || [];
+            devices.map(storeDev.bind(null, u));
+        }));
+    });
+
+    return q.all(promises).then(function() {
+        return stored;
+    });
+};
+
+/**
+ * @param {string[]} downloadUsers list of userIds
+ *
+ * @return {Object a map from userId to a promise for a result for that user
+ */
+Crypto.prototype._doKeyDownloadForUsers = function(downloadUsers) {
+    var self = this;
+
+    console.log('Starting key download for ' + downloadUsers);
+
+    var deferMap = {};
+    var promiseMap = {};
+
+    downloadUsers.map(function(u) {
+        deferMap[u] = q.defer();
+        promiseMap[u] = deferMap[u].promise;
+    });
+
+    this._baseApis.downloadKeysForUsers(
         downloadUsers
-    ).then(function(res) {
+    ).done(function(res) {
         var dk = res.device_keys || {};
 
         for (var i = 0; i < downloadUsers.length; ++i) {
             var userId = downloadUsers[i];
-            // console.log('keys for ' + userId + ':', dk[userId]);
+            var deviceId;
+
+            console.log('got keys for ' + userId + ':', dk[userId]);
 
             if (!dk[userId]) {
                 // no result for this user
-                // TODO: do something with failures
+                var err = 'Unknown';
+                // TODO: do something with res.failures
+                deferMap[userId].reject(err);
                 continue;
             }
 
             // map from deviceid -> deviceinfo for this user
-            var userStore = stored[userId];
+            var userStore = {};
+            var devs = self._sessionStore.getEndToEndDevicesForUser(userId);
+            if (devs) {
+                for (deviceId in devs) {
+                    if (devs.hasOwnProperty(deviceId)) {
+                        var d = DeviceInfo.fromStorage(devs[deviceId], deviceId);
+                        userStore[deviceId] = d;
+                    }
+                }
+            }
+
             _updateStoredDeviceKeysForUser(
                 self._olmDevice, userId, userStore, dk[userId]
             );
 
             // update the session store
             var storage = {};
-            for (var deviceId in userStore) {
+            for (deviceId in userStore) {
                 if (!userStore.hasOwnProperty(deviceId)) {
                     continue;
                 }
@@ -331,9 +385,16 @@ Crypto.prototype.downloadKeys = function(userIds, forceDownload) {
             self._sessionStore.storeEndToEndDevicesForUser(
                 userId, storage
             );
+
+            deferMap[userId].resolve();
         }
-        return stored;
+    }, function(err) {
+        downloadUsers.map(function(u) {
+            deferMap[u].reject(err);
+        });
     });
+
+    return promiseMap;
 };
 
 function _updateStoredDeviceKeysForUser(_olmDevice, userId, userStore,
@@ -462,6 +523,22 @@ Crypto.prototype.getStoredDevicesForUser = function(userId) {
     return res;
 };
 
+/**
+ * Get the stored keys for a single device
+ *
+ * @param {string} userId
+ * @param {string} deviceId
+ *
+ * @return {module:crypto/deviceinfo?} list of devices, or undefined
+ * if we don't know about this device
+ */
+Crypto.prototype.getStoredDevice = function(userId, deviceId) {
+    var devs = this._sessionStore.getEndToEndDevicesForUser(userId);
+    if (!devs || !devs[deviceId]) {
+        return undefined;
+    }
+    return DeviceInfo.fromStorage(devs[deviceId], deviceId);
+};
 
 /**
  * List the stored device keys for a user id
@@ -998,6 +1075,11 @@ Crypto.prototype._onCryptoEvent = function(event) {
  * @param {module:models/room[]} rooms list of rooms the client knows about
  */
 Crypto.prototype._onInitialSyncCompleted = function(rooms) {
+    this._initialSyncCompleted = true;
+
+    // catch up on any m.new_device events which arrived during the initial sync.
+    this._flushNewDeviceRequests();
+
     if (this._sessionStore.getDeviceAnnounced()) {
         return;
     }
@@ -1124,26 +1206,58 @@ Crypto.prototype._onNewDeviceEvent = function(event) {
     console.log("m.new_device event from " + userId + ":" + deviceId +
                 " for rooms " + rooms);
 
+    if (this.getStoredDevice(userId, deviceId)) {
+        console.log("Known device; ignoring");
+        return;
+    }
+
+    this._pendingNewDevices[userId] = this._pendingNewDevices[userId] || {};
+    this._pendingNewDevices[userId][deviceId] = true;
+
+    // we delay handling these until the intialsync has completed, so that we
+    // can do all of them together.
+    if (this._initialSyncCompleted) {
+        this._flushNewDeviceRequests();
+    }
+};
+
+/**
+ * Start device queries for any users who sent us an m.new_device recently
+ */
+Crypto.prototype._flushNewDeviceRequests = function() {
     var self = this;
-    this.downloadKeys(
+
-        [userId], true
+    var pending = this._pendingNewDevices;
-    ).then(function() {
+    var users = utils.keys(pending).filter(function(u) {
-        for (var i = 0; i < rooms.length; i++) {
+        return utils.keys(pending[u]).length > 0;
-            var roomId = rooms[i];
+    });
-            var alg = self._roomEncryptors[roomId];
+
-            if (!alg) {
+    if (users.length === 0) {
-                // not encrypting in this room
+        return;
-                continue;
+    }
-            }
+
-            alg.onNewDevice(userId, deviceId);
+    var r = this._doKeyDownloadForUsers(users);
-        }
+
-    }).catch(function(e) {
+    // we've kicked off requests to these users: remove their
-        console.error(
+    // pending flag for now.
-            "Error updating device keys for new device " + userId + ":" +
+    this._pendingNewDevices = {};
-                deviceId,
+
-            e
+    users.map(function(u) {
-        );
+        r[u] = r[u].catch(function(e) {
-    }).done();
+            console.error(
+                'Error updating device keys for user ' + u + ':', e
+            );
+
+            // reinstate the pending flags on any users which failed; this will
+            // mean that we will do another download in the future, but won't
+            // tight-loop.
+            //
+            self._pendingNewDevices[u] = self._pendingNewDevices[u] || {};
+            utils.update(self._pendingNewDevices[u], pending[u]);
+        });
+    });
+
+    q.all(utils.values(r)).done();
 };
 
 /**

spec/integ/matrix-client-crypto.spec.js

@@ -739,4 +739,31 @@ describe("MatrixClient crypto", function() {
             }).then(aliRecvMessage)
             .catch(test_utils.failTest).done(done);
     });
+
+
+    it("Ali does a key query when she gets a new_device event", function(done) {
+        q()
+            .then(bobUploadsKeys)
+            .then(aliStartClient)
+            .then(function() {
+                var syncData = {
+                    next_batch: '2',
+                    to_device: {
+                        events: [
+                            test_utils.mkEvent({
+                                content: {
+                                    device_id: 'TEST_DEVICE',
+                                    rooms: [],
+                                },
+                                sender: bobUserId,
+                                type: 'm.new_device',
+                            }),
+                        ],
+                    },
+                };
+                aliHttpBackend.when('GET', '/sync').respond(200, syncData);
+                return aliHttpBackend.flush('/sync', 1);
+            }).then(expectAliQueryKeys)
+            .nodeify(done);
+    });
 });

spec/integ/matrix-client-methods.spec.js

@@ -371,21 +371,6 @@ describe("MatrixClient", function() {
 
             httpBackend.flush();
         });
-
-        it("should return a rejected promise if the request fails", function(done) {
-            httpBackend.when("POST", "/keys/query").respond(400);
-
-            var exceptionThrown;
-            client.downloadKeys(["bottom"]).then(function() {
-                fail("download didn't fail");
-            }, function(err) {
-                exceptionThrown = err;
-            }).then(function() {
-                expect(exceptionThrown).toBeTruthy();
-            }).catch(utils.failTest).done(done);
-
-            httpBackend.flush();
-        });
     });
 
     describe("deleteDevice", function() {