Make sliding sync linearize processing of sync requests (#3442)

* Make sliding sync linearize processing of sync requests

* Iterate

* Iterate

* Iterate

* Iterate

spec/integ/sliding-sync-sdk.spec.ts

@@ -20,7 +20,7 @@ import { fail } from "assert";
 
 import { SlidingSync, SlidingSyncEvent, MSC3575RoomData, SlidingSyncState, Extension } from "../../src/sliding-sync";
 import { TestClient } from "../TestClient";
-import { IRoomEvent, IStateEvent } from "../../src/sync-accumulator";
+import { IRoomEvent, IStateEvent } from "../../src";
 import {
     MatrixClient,
     MatrixEvent,
@@ -39,7 +39,7 @@ import {
 } from "../../src";
 import { SlidingSyncSdk } from "../../src/sliding-sync-sdk";
 import { SyncApiOptions, SyncState } from "../../src/sync";
-import { IStoredClientOpts } from "../../src/client";
+import { IStoredClientOpts } from "../../src";
 import { logger } from "../../src/logger";
 import { emitPromise } from "../test-utils/test-utils";
 import { defer } from "../../src/utils";

src/models/typed-event-emitter.ts

@@ -89,6 +89,24 @@ export class TypedEventEmitter<
         return super.emit(event, ...args);
     }
 
+    /**
+     * Similar to `emit` but calls all listeners within a `Promise.all` and returns the promise chain
+     * @param event - The name of the event to emit
+     * @param args - Arguments to pass to the listener
+     * @returns `true` if the event had listeners, `false` otherwise.
+     */
+    public async emitPromised<T extends Events>(
+        event: T,
+        ...args: Parameters<SuperclassArguments[T]>
+    ): Promise<boolean>;
+    public async emitPromised<T extends Events>(event: T, ...args: Parameters<Arguments[T]>): Promise<boolean>;
+    public async emitPromised<T extends Events>(event: T, ...args: any[]): Promise<boolean> {
+        const listeners = this.listeners(event);
+        return Promise.allSettled(listeners.map((l) => l(...args))).then(() => {
+            return listeners.length > 0;
+        });
+    }
+
     /**
      * Returns the number of listeners listening to the event named `event`.
      *

src/sliding-sync-sdk.ts

@@ -376,7 +376,7 @@ export class SlidingSyncSdk {
         });
     }
 
-    private onRoomData(roomId: string, roomData: MSC3575RoomData): void {
+    private async onRoomData(roomId: string, roomData: MSC3575RoomData): Promise<void> {
         let room = this.client.store.getRoom(roomId);
         if (!room) {
             if (!roomData.initial) {
@@ -385,7 +385,7 @@ export class SlidingSyncSdk {
             }
             room = _createAndReEmitRoom(this.client, roomId, this.opts);
         }
-        this.processRoomData(this.client, room, roomData);
+        await this.processRoomData(this.client, room!, roomData);
     }
 
     private onLifecycle(state: SlidingSyncState, resp: MSC3575SlidingSyncResponse | null, err?: Error): void {

src/sliding-sync.ts

@@ -326,7 +326,7 @@ export enum SlidingSyncEvent {
 }
 
 export type SlidingSyncEventHandlerMap = {
-    [SlidingSyncEvent.RoomData]: (roomId: string, roomData: MSC3575RoomData) => void;
+    [SlidingSyncEvent.RoomData]: (roomId: string, roomData: MSC3575RoomData) => Promise<void> | void;
     [SlidingSyncEvent.Lifecycle]: (
         state: SlidingSyncState,
         resp: MSC3575SlidingSyncResponse | null,
@@ -567,14 +567,14 @@ export class SlidingSync extends TypedEventEmitter<SlidingSyncEvent, SlidingSync
      * @param roomId - The room which received some data.
      * @param roomData - The raw sliding sync response JSON.
      */
-    private invokeRoomDataListeners(roomId: string, roomData: MSC3575RoomData): void {
+    private async invokeRoomDataListeners(roomId: string, roomData: MSC3575RoomData): Promise<void> {
         if (!roomData.required_state) {
             roomData.required_state = [];
         }
         if (!roomData.timeline) {
             roomData.timeline = [];
         }
-        this.emit(SlidingSyncEvent.RoomData, roomId, roomData);
+        await this.emitPromised(SlidingSyncEvent.RoomData, roomId, roomData);
     }
 
     /**
@@ -923,9 +923,9 @@ export class SlidingSync extends TypedEventEmitter<SlidingSyncEvent, SlidingSync
             }
             this.onPreExtensionsResponse(resp.extensions);
 
-            Object.keys(resp.rooms).forEach((roomId) => {
-                this.invokeRoomDataListeners(roomId, resp!.rooms[roomId]);
-            });
+            for (const roomId in resp.rooms) {
+                await this.invokeRoomDataListeners(roomId, resp!.rooms[roomId]);
+            }
 
             const listKeysWithUpdates: Set<string> = new Set();
             if (!doNotUpdateList) {