move key backup operation and flush bookkeeping inside KeyBackup

so we can flush from other places than Session

src/matrix/Session.js

@@ -76,7 +76,6 @@ export class Session {
         this._getSyncToken = () => this.syncToken;
         this._olmWorker = olmWorker;
         this._keyBackup = null;
-        this._keyBackupOperation = new ObservableValue(null);
         this._hasSecretStorageKey = new ObservableValue(null);
         this._observedRoomStatus = new Map();
 
@@ -135,15 +134,16 @@ export class Session {
             olmUtil: this._olmUtil,
             senderKeyLock
         });
+        this._keyLoader = new MegOlmKeyLoader(this._olm, PICKLE_KEY, 20);
         this._megolmEncryption = new MegOlmEncryption({
             account: this._e2eeAccount,
             pickleKey: PICKLE_KEY,
             olm: this._olm,
             storage: this._storage,
+            keyLoader: this._keyLoader,
             now: this._platform.clock.now,
             ownDeviceId: this._sessionInfo.deviceId,
         });
-        this._keyLoader = new MegOlmKeyLoader(this._olm, PICKLE_KEY, 20);
         this._megolmDecryption = new MegOlmDecryption(this._keyLoader, this._olmWorker);
         this._deviceMessageHandler.enableEncryption({olmDecryption, megolmDecryption: this._megolmDecryption});
     }
@@ -282,10 +282,6 @@ export class Session {
         return this._keyBackup;
     }
 
-    get keyBackupOperation() {
-        return this._keyBackupOperation;
-    }
-
     get hasIdentity() {
         return !!this._e2eeAccount;
     }
@@ -635,23 +631,8 @@ export class Session {
                 await log.wrap("uploadKeys", log => this._e2eeAccount.uploadKeys(this._storage, false, log));
             }
         }
-        // should flush and not already flushing
+        if (changes.hasNewRoomKeys) {
-        if (changes.hasNewRoomKeys && this._keyBackup && !this._keyBackupOperation.get()) {
+            this._keyBackup?.flush(log);
-            log.wrapDetached("flush key backup", async log => {
-                const operation = this._keyBackup.flush(log);
-                this._keyBackupOperation.set(operation);
-                try {
-                    const success = await operation.result;
-                    // stop key backup if the version was changed
-                    if (!success) {
-                        this._keyBackup = this._keyBackup.dispose();
-                        this.needsKeyBackup.set(true);
-                    }
-                } catch (err) {
-                    log.catch(err);
-                }
-                this._keyBackupOperation.set(null);
-            });
         }
     }
 

src/matrix/e2ee/megolm/keybackup/KeyBackup.ts

@@ -19,6 +19,7 @@ import {StoredRoomKey, keyFromBackup} from "../decryption/RoomKey";
 import {MEGOLM_ALGORITHM} from "../../common";
 import * as Curve25519 from "./Curve25519";
 import {AbortableOperation} from "../../../../utils/AbortableOperation";
+import {ObservableValue} from "../../../../observable/ObservableValue";
 
 import {SetAbortableFn} from "../../../../utils/AbortableOperation";
 import type {BackupInfo, SessionData, SessionKeyInfo, SessionInfo, KeyBackupPayload} from "./types";
@@ -33,7 +34,12 @@ import type {Transaction} from "../../../storage/idb/Transaction";
 import type * as OlmNamespace from "@matrix-org/olm";
 type Olm = typeof OlmNamespace;
 
+const KEYS_PER_REQUEST = 20;
+
 export class KeyBackup {
+    public readonly operationInProgress = new ObservableValue<AbortableOperation<Promise<boolean>, Progress> | undefined>(undefined);
+    public readonly needsNewKey = new ObservableValue(false);
+
     constructor(
         private readonly backupInfo: BackupInfo,
         private readonly crypto: Curve25519.BackupEncryption,
@@ -56,13 +62,32 @@ export class KeyBackup {
         }
     }
 
-    // TODO: protect against having multiple concurrent flushes
+    flush(log: ILogItem): void {
-    flush(log: ILogItem): AbortableOperation<Promise<boolean>, Progress> {
+        if (!this.operationInProgress.get()) {
+            log.wrapDetached("flush key backup", async log => {
+                const operation = this._flush(log);
+                this.operationInProgress.set(operation);
+                try {
+                    const success = await operation.result;
+                    // stop key backup if the version was changed
+                    if (!success) {
+                        this.needsNewKey.set(true);
+                    }
+                } catch (err) {
+                    log.catch(err);
+                }
+                this.operationInProgress.set(undefined);
+            });
+        }
+    }
+
+    private _flush(log: ILogItem): AbortableOperation<Promise<boolean>, Progress> {
         return new AbortableOperation(async (setAbortable, setProgress) => {
             let total = 0;
             let amountFinished = 0;
             while (true) {
-                const timeout = this.platform.clock.createTimeout(this.platform.random() * 10000);
+                const waitMs = this.platform.random() * 10000;
+                const timeout = this.platform.clock.createTimeout(waitMs);
                 setAbortable(timeout);
                 await timeout.elapsed();
                 const txn = await this.storage.readTxn([StoreNames.inboundGroupSessions]);
@@ -70,7 +95,7 @@ export class KeyBackup {
                 // fetch total again on each iteration as while we are flushing, sync might be adding keys
                 total = await txn.inboundGroupSessions.countNonBackedUpSessions();
                 setProgress(new Progress(total, amountFinished));
-                const keysNeedingBackup = (await txn.inboundGroupSessions.getFirstNonBackedUpSessions(20))
+                const keysNeedingBackup = (await txn.inboundGroupSessions.getFirstNonBackedUpSessions(KEYS_PER_REQUEST))
                     .map(entry => new StoredRoomKey(entry));
                 if (keysNeedingBackup.length === 0) {
                     return true;
@@ -114,7 +139,9 @@ export class KeyBackup {
         ]);
         setAbortable(txn);
         try {
-            await Promise.all(roomKeys.map(key => txn.inboundGroupSessions.markAsBackedUp(key.roomId, key.senderKey, key.sessionId)));
+            await Promise.all(roomKeys.map(key => {
+                return txn.inboundGroupSessions.markAsBackedUp(key.roomId, key.senderKey, key.sessionId);
+            }));
         } catch (err) {
             txn.abort();
             throw err;