Resume to-device message queue after resumed sync

spec/unit/ToDeviceMessageQueue.spec.ts

@@ -0,0 +1,106 @@
+import { ConnectionError } from "../../src/http-api/errors";
+import { ClientEvent, MatrixClient, Store } from "../../src/client";
+import { ToDeviceMessageQueue } from "../../src/ToDeviceMessageQueue";
+import { getMockClientWithEventEmitter } from "../test-utils/client";
+import { StubStore } from "../../src/store/stub";
+import { IndexedToDeviceBatch } from "../../src/models/ToDeviceMessage";
+import { SyncState } from "../../src/sync";
+
+describe("onResumedSync", () => {
+    let batch: IndexedToDeviceBatch | null;
+    let shouldFailSendToDevice: Boolean;
+    let onSendToDeviceFailure: () => void;
+    let onSendToDeviceSuccess: () => void;
+    let resumeSync: (newState: SyncState, oldState: SyncState) => void;
+
+    let store: Store;
+    let mockClient: MatrixClient;
+    let queue: ToDeviceMessageQueue;
+
+    beforeEach(() => {
+        batch = {
+            id: 0,
+            txnId: "123",
+            eventType: "m.dummy",
+            batch: [],
+        };
+
+        shouldFailSendToDevice = true;
+        onSendToDeviceFailure = () => {};
+        onSendToDeviceSuccess = () => {};
+        resumeSync = (newState, oldState) => {
+            shouldFailSendToDevice = false;
+            mockClient.emit(ClientEvent.Sync, newState, oldState);
+        };
+
+        store = new StubStore();
+        store.getOldestToDeviceBatch = jest.fn().mockImplementation(() => {
+            return batch;
+        });
+        store.removeToDeviceBatch = jest.fn().mockImplementation(() => {
+            batch = null;
+        });
+
+        mockClient = getMockClientWithEventEmitter({});
+        mockClient.store = store;
+        mockClient.sendToDevice = jest.fn().mockImplementation(async () => {
+            if (shouldFailSendToDevice) {
+                await Promise.reject(new ConnectionError("")).finally(() => {
+                    setTimeout(onSendToDeviceFailure, 0);
+                });
+            } else {
+                await Promise.resolve({}).finally(() => {
+                    setTimeout(onSendToDeviceSuccess, 0);
+                });
+            }
+        });
+
+        queue = new ToDeviceMessageQueue(mockClient);
+    });
+
+    it("resends queue after connectivity restored", (done) => {
+        onSendToDeviceFailure = () => {
+            expect(store.getOldestToDeviceBatch).toHaveBeenCalledTimes(1);
+            expect(store.removeToDeviceBatch).not.toHaveBeenCalled();
+
+            resumeSync(SyncState.Syncing, SyncState.Catchup);
+            expect(store.getOldestToDeviceBatch).toHaveBeenCalledTimes(2);
+        };
+
+        onSendToDeviceSuccess = () => {
+            expect(store.getOldestToDeviceBatch).toHaveBeenCalledTimes(3);
+            expect(store.removeToDeviceBatch).toHaveBeenCalled();
+            done();
+        };
+
+        queue.start();
+    });
+
+    it("does not resend queue if client sync still catching up", (done) => {
+        onSendToDeviceFailure = () => {
+            expect(store.getOldestToDeviceBatch).toHaveBeenCalledTimes(1);
+            expect(store.removeToDeviceBatch).not.toHaveBeenCalled();
+
+            resumeSync(SyncState.Catchup, SyncState.Catchup);
+            expect(store.getOldestToDeviceBatch).toHaveBeenCalledTimes(1);
+            done();
+        };
+
+        queue.start();
+    });
+
+    it("does not resend queue if connectivity restored after queue stopped", (done) => {
+        onSendToDeviceFailure = () => {
+            expect(store.getOldestToDeviceBatch).toHaveBeenCalledTimes(1);
+            expect(store.removeToDeviceBatch).not.toHaveBeenCalled();
+
+            queue.stop();
+
+            resumeSync(SyncState.Syncing, SyncState.Catchup);
+            expect(store.getOldestToDeviceBatch).toHaveBeenCalledTimes(1);
+            done();
+        };
+
+        queue.start();
+    });
+});

src/ToDeviceMessageQueue.ts

@@ -16,9 +16,11 @@ limitations under the License.
 
 import { ToDeviceMessageId } from "./@types/event";
 import { logger } from "./logger";
-import { MatrixError, MatrixClient } from "./matrix";
+import { MatrixClient, ClientEvent } from "./client";
+import { MatrixError } from "./http-api";
 import { IndexedToDeviceBatch, ToDeviceBatch, ToDeviceBatchWithTxnId, ToDevicePayload } from "./models/ToDeviceMessage";
 import { MatrixScheduler } from "./scheduler";
+import { SyncState } from "./sync";
 
 const MAX_BATCH_SIZE = 20;
 
@@ -37,12 +39,14 @@ export class ToDeviceMessageQueue {
     public start(): void {
         this.running = true;
         this.sendQueue();
+        this.client.on(ClientEvent.Sync, this.onResumedSync);
     }
 
     public stop(): void {
         this.running = false;
         if (this.retryTimeout !== null) clearTimeout(this.retryTimeout);
         this.retryTimeout = null;
+        this.client.removeListener(ClientEvent.Sync, this.onResumedSync);
     }
 
     public async queueBatch(batch: ToDeviceBatch): Promise<void> {
@@ -132,4 +136,15 @@ export class ToDeviceMessageQueue {
 
         await this.client.sendToDevice(batch.eventType, contentMap, batch.txnId);
     }
+
+    /**
+     * Listen to sync state changes and automatically resend any pending events
+     * once syncing is resumed
+     */
+    private onResumedSync = (state: SyncState | null, oldState: SyncState | null): void => {
+        if (state === SyncState.Syncing && oldState !== SyncState.Syncing) {
+            logger.info(`Resuming queue after resumed sync`);
+            this.sendQueue();
+        }
+    };
 }