Make Crypto._processReceivedRoomKeyRequests async

This is slightly complicated by the fact that it's initiated from a synchronous
process which we don't want to make async (processing the /sync response) and
we want to avoid racing two copies of the processor.

src/crypto/index.js

@@ -104,6 +104,8 @@ function Crypto(baseApis, sessionStore, userId, deviceId,
     // we received in the current sync.
     this._receivedRoomKeyRequests = [];
     this._receivedRoomKeyRequestCancellations = [];
+    // true if we are currently processing received room key requests
+    this._processingRoomKeyRequests = false;
 
     let myDevices = this._sessionStore.getEndToEndDevicesForUser(
         this._userId,
@@ -1059,10 +1061,50 @@ Crypto.prototype._onRoomKeyRequestEvent = function(event) {
  *
  * @private
  */
-Crypto.prototype._processReceivedRoomKeyRequests = function() {
+Crypto.prototype._processReceivedRoomKeyRequests = async function() {
+    if (this._processingRoomKeyRequests) {
+        // we're still processing last time's requests; keep queuing new ones
+        // up for now.
+        return;
+    }
+    this._processingRoomKeyRequests = true;
+
+    try {
+        // we need to grab and clear the queues in the synchronous bit of this method,
+        // so that we don't end up racing with the next /sync.
         const requests = this._receivedRoomKeyRequests;
         this._receivedRoomKeyRequests = [];
-    for (const req of requests) {
+        const cancellations = this._receivedRoomKeyRequestCancellations;
+        this._receivedRoomKeyRequestCancellations = [];
+
+        // Process all of the requests, *then* all of the cancellations.
+        //
+        // This makes sure that if we get a request and its cancellation in the
+        // same /sync result, then we process the request before the
+        // cancellation (and end up with a cancelled request), rather than the
+        // cancellation before the request (and end up with an outstanding
+        // request which should have been cancelled.)
+        await Promise.map(
+            requests, (req) =>
+                this._processReceivedRoomKeyRequest(req),
+        );
+        await Promise.map(
+            cancellations, (cancellation) =>
+                this._processReceivedRoomKeyRequestCancellation(cancellation),
+        );
+    } catch (e) {
+        console.error(`Error processing room key requsts: ${e}`);
+    } finally {
+        this._processingRoomKeyRequests = false;
+    }
+};
+
+/**
+ * Helper for processReceivedRoomKeyRequests
+ *
+ * @param {IncomingRoomKeyRequest} req
+ */
+Crypto.prototype._processReceivedRoomKeyRequest = async function(req) {
     const userId = req.userId;
     const deviceId = req.deviceId;
 
@@ -1087,13 +1129,13 @@ Crypto.prototype._processReceivedRoomKeyRequests = function() {
     // the keys for the requested events, and can drop the requests.
     if (!this._roomDecryptors[roomId]) {
         console.log(`room key request for unencrypted room ${roomId}`);
-            continue;
+        return;
     }
 
     const decryptor = this._roomDecryptors[roomId][alg];
     if (!decryptor) {
         console.log(`room key request for unknown alg ${alg} in room ${roomId}`);
-            continue;
+        return;
     }
 
     if (!decryptor.hasKeysForKeyRequest(req)) {
@@ -1101,7 +1143,7 @@ Crypto.prototype._processReceivedRoomKeyRequests = function() {
             `room key request for unknown session ${roomId} / ` +
                 body.session_id,
         );
-            continue;
+        return;
     }
 
     req.share = () => {
@@ -1117,11 +1159,17 @@ Crypto.prototype._processReceivedRoomKeyRequests = function() {
     }
 
     this.emit("crypto.roomKeyRequest", req);
-    }
+};
 
-    const cancellations = this._receivedRoomKeyRequestCancellations;
-    this._receivedRoomKeyRequestCancellations = [];
-    for (const cancellation of cancellations) {
+
+/**
+ * Helper for processReceivedRoomKeyRequests
+ *
+ * @param {IncomingRoomKeyRequestCancellation} cancellation
+ */
+Crypto.prototype._processReceivedRoomKeyRequestCancellation = async function(
+    cancellation,
+) {
     console.log(
         `m.room_key_request cancellation for ${cancellation.userId}:` +
             `${cancellation.deviceId} (id ${cancellation.requestId})`,
@@ -1131,7 +1179,6 @@ Crypto.prototype._processReceivedRoomKeyRequests = function() {
     // about, but we don't currently have a record of that, so we just pass
     // everything through.
     this.emit("crypto.roomKeyRequestCancellation", cancellation);
-    }
 };
 
 /**