Support cancelling events whilst they are in status = ENCRYPTING (#2095)

Author: t3chguy

spec/test-utils.js

@@ -365,3 +365,5 @@ export function setHttpResponses(
             .respond(200, response.data);
     });
 }
+
+export const emitPromise = (e, k) => new Promise(r => e.once(k, r));

spec/unit/matrix-client.spec.js

@@ -11,8 +11,9 @@ import {
     UNSTABLE_MSC3089_TREE_SUBTYPE,
 } from "../../src/@types/event";
 import { MEGOLM_ALGORITHM } from "../../src/crypto/olmlib";
-import { MatrixEvent } from "../../src/models/event";
+import { EventStatus, MatrixEvent } from "../../src/models/event";
 import { Preset } from "../../src/@types/partials";
+import * as testUtils from "../test-utils";
 
 jest.useFakeTimers();
 
@@ -867,4 +868,83 @@ describe("MatrixClient", function() {
             await client.redactEvent(roomId, eventId, txnId, { reason });
         });
     });
+
+    describe("cancelPendingEvent", () => {
+        const roomId = "!room:server";
+        const txnId = "m12345";
+
+        const mockRoom = {
+            getMyMembership: () => "join",
+            updatePendingEvent: (event, status) => event.setStatus(status),
+            currentState: {
+                getStateEvents: (eventType, stateKey) => {
+                    if (eventType === EventType.RoomCreate) {
+                        expect(stateKey).toEqual("");
+                        return new MatrixEvent({
+                            content: {
+                                [RoomCreateTypeField]: RoomType.Space,
+                            },
+                        });
+                    } else if (eventType === EventType.RoomEncryption) {
+                        expect(stateKey).toEqual("");
+                        return new MatrixEvent({ content: {} });
+                    } else {
+                        throw new Error("Unexpected event type or state key");
+                    }
+                },
+            },
+        };
+
+        let event;
+        beforeEach(async () => {
+            event = new MatrixEvent({
+                event_id: "~" + roomId + ":" + txnId,
+                user_id: client.credentials.userId,
+                sender: client.credentials.userId,
+                room_id: roomId,
+                origin_server_ts: new Date().getTime(),
+            });
+            event.setTxnId(txnId);
+
+            client.getRoom = (getRoomId) => {
+                expect(getRoomId).toEqual(roomId);
+                return mockRoom;
+            };
+            client.crypto = { // mock crypto
+                encryptEvent: (event, room) => new Promise(() => {}),
+            };
+        });
+
+        function assertCancelled() {
+            expect(event.status).toBe(EventStatus.CANCELLED);
+            expect(client.scheduler.removeEventFromQueue(event)).toBeFalsy();
+            expect(httpLookups.filter(h => h.path.includes("/send/")).length).toBe(0);
+        }
+
+        it("should cancel an event which is queued", () => {
+            event.setStatus(EventStatus.QUEUED);
+            client.scheduler.queueEvent(event);
+            client.cancelPendingEvent(event);
+            assertCancelled();
+        });
+
+        it("should cancel an event which is encrypting", async () => {
+            client.encryptAndSendEvent(null, event);
+            await testUtils.emitPromise(event, "Event.status");
+            client.cancelPendingEvent(event);
+            assertCancelled();
+        });
+
+        it("should cancel an event which is not sent", () => {
+            event.setStatus(EventStatus.NOT_SENT);
+            client.cancelPendingEvent(event);
+            assertCancelled();
+        });
+
+        it("should error when given any other event status", () => {
+            event.setStatus(EventStatus.SENDING);
+            expect(() => client.cancelPendingEvent(event)).toThrow("cannot cancel an event with status sending");
+            expect(event.status).toBe(EventStatus.SENDING);
+        });
+    });
 });

src/client.ts

@@ -21,7 +21,7 @@ limitations under the License.
 
 import { EventEmitter } from "events";
 
-import { ISyncStateData, SyncApi } from "./sync";
+import { ISyncStateData, SyncApi, SyncState } from "./sync";
 import { EventStatus, IContent, IDecryptOptions, IEvent, MatrixEvent } from "./models/event";
 import { StubStore } from "./store/stub";
 import { createNewMatrixCall, MatrixCall } from "./webrtc/call";
@@ -96,7 +96,6 @@ import {
     IRecoveryKey,
     ISecretStorageKeyInfo,
 } from "./crypto/api";
-import { SyncState } from "./sync";
 import { EventTimelineSet } from "./models/event-timeline-set";
 import { VerificationRequest } from "./crypto/verification/request/VerificationRequest";
 import { VerificationBase as Verification } from "./crypto/verification/Base";
@@ -816,6 +815,7 @@ export class MatrixClient extends EventEmitter {
     protected exportedOlmDeviceToImport: IOlmDevice;
     protected txnCtr = 0;
     protected mediaHandler = new MediaHandler();
+    protected pendingEventEncryption = new Map<string, Promise<void>>();
 
     constructor(opts: IMatrixClientCreateOpts) {
         super();
@@ -870,7 +870,7 @@ export class MatrixClient extends EventEmitter {
 
         this.scheduler = opts.scheduler;
         if (this.scheduler) {
-            this.scheduler.setProcessFunction(async (eventToSend) => {
+            this.scheduler.setProcessFunction(async (eventToSend: MatrixEvent) => {
                 const room = this.getRoom(eventToSend.getRoomId());
                 if (eventToSend.status !== EventStatus.SENDING) {
                     this.updatePendingEventStatus(room, eventToSend, EventStatus.SENDING);
@@ -3399,15 +3399,18 @@ export class MatrixClient extends EventEmitter {
      * Cancel a queued or unsent event.
      *
      * @param {MatrixEvent} event   Event to cancel
-     * @throws Error if the event is not in QUEUED or NOT_SENT state
+     * @throws Error if the event is not in QUEUED, NOT_SENT or ENCRYPTING state
      */
     public cancelPendingEvent(event: MatrixEvent) {
-        if ([EventStatus.QUEUED, EventStatus.NOT_SENT].indexOf(event.status) < 0) {
+        if (![EventStatus.QUEUED, EventStatus.NOT_SENT, EventStatus.ENCRYPTING].includes(event.status)) {
             throw new Error("cannot cancel an event with status " + event.status);
         }
 
-        // first tell the scheduler to forget about it, if it's queued
-        if (this.scheduler) {
+        // if the event is currently being encrypted then
+        if (event.status === EventStatus.ENCRYPTING) {
+            this.pendingEventEncryption.delete(event.getId());
+        } else if (this.scheduler && event.status === EventStatus.QUEUED) {
+            // tell the scheduler to forget about it, if it's queued
             this.scheduler.removeEventFromQueue(event);
         }
 
@@ -3669,16 +3672,26 @@ export class MatrixClient extends EventEmitter {
      * @private
      */
     private encryptAndSendEvent(room: Room, event: MatrixEvent, callback?: Callback): Promise<ISendEventResponse> {
+        let cancelled = false;
         // Add an extra Promise.resolve() to turn synchronous exceptions into promise rejections,
         // so that we can handle synchronous and asynchronous exceptions with the
         // same code path.
         return Promise.resolve().then(() => {
             const encryptionPromise = this.encryptEventIfNeeded(event, room);
-            if (!encryptionPromise) return null;
+            if (!encryptionPromise) return null; // doesn't need encryption
 
+            this.pendingEventEncryption.set(event.getId(), encryptionPromise);
             this.updatePendingEventStatus(room, event, EventStatus.ENCRYPTING);
-            return encryptionPromise.then(() => this.updatePendingEventStatus(room, event, EventStatus.SENDING));
+            return encryptionPromise.then(() => {
+                if (!this.pendingEventEncryption.has(event.getId())) {
+                    // cancelled via MatrixClient::cancelPendingEvent
+                    cancelled = true;
+                    return;
+                }
+                this.updatePendingEventStatus(room, event, EventStatus.SENDING);
+            });
         }).then(() => {
+            if (cancelled) return {} as ISendEventResponse;
             let promise: Promise<ISendEventResponse>;
             if (this.scheduler) {
                 // if this returns a promise then the scheduler has control now and will

src/crypto/index.ts

@@ -2728,7 +2728,6 @@ export class Crypto extends EventEmitter {
         }
     }
 
-    /* eslint-disable valid-jsdoc */    //https://github.com/eslint/eslint/issues/7307
     /**
      * Encrypt an event according to the configuration of the room.
      *
@@ -2739,8 +2738,6 @@ export class Crypto extends EventEmitter {
      * @return {Promise?} Promise which resolves when the event has been
      *     encrypted, or null if nothing was needed
      */
-    /* eslint-enable valid-jsdoc */
-    // TODO this return type lies
     public async encryptEvent(event: MatrixEvent, room: Room): Promise<void> {
         if (!room) {
             throw new Error("Cannot send encrypted messages in unknown rooms");

src/models/room.ts

@@ -2302,12 +2302,12 @@ function pendingEventsKey(roomId: string): string {
     return `mx_pending_events_${roomId}`;
 }
 
-/* a map from current event status to a list of allowed next statuses
-     */
+// a map from current event status to a list of allowed next statuses
 const ALLOWED_TRANSITIONS: Record<EventStatus, EventStatus[]> = {
     [EventStatus.ENCRYPTING]: [
         EventStatus.SENDING,
         EventStatus.NOT_SENT,
+        EventStatus.CANCELLED,
     ],
     [EventStatus.SENDING]: [
         EventStatus.ENCRYPTING,