MatrixRTC: New membership manager (#4726)

* WIP doodles on MembershipManager test cases

* .

* initial membership manager test setup.

* Updates from discussion

* revert renaming comments

* remove unused import

* fix leave delayed event resend test.
It was missing a flush.

* comment out and remove unused variables

* es lint

* use jsdom instead of node test environment

* remove unused variables

* remove unused export

* temp

* review

* fixup tests

* more review

* remove wait for expect dependency

* temp

* fix wrong mocked meberhsip template

* rename MembershipManager -> LegacyMembershipManager
And remove the IMembershipManager from it

* Add new memberhsip manager

* fix tests to be compatible with old and new membership manager

* Comment cleanup

* Allow join to throw
 - Add tests for throwing cases
 - Fixs based on tests

* introduce membershipExpiryTimeoutSlack

* more detailed comments and cleanup

* warn if slack is misconfigured and use default values instead

* fix action resets.

* flatten MembershipManager.spec.ts

* rename testEnvironment to memberManagerTestEnvironment

* allow configuring Legacy manager in the matrixRTC session

* deprecate LegacyMembershipManager

* remove usage of waitForExpect

* flatten tests and add comments

* clean up leave logic branch

* add more leave test cases

* use defer

* review ("Some minor tidying things for now.")

* add onError for join method and cleanup

* use pop instead of filter

* fixes

* simplify error handling and MembershipAction
Only use one membership action enum

* Add diagram

* fix new error api in rtc session

* fix up retry counter

* fix lints

* make unrecoverable errors more explicit

* fix tests

* Allow multiple retries on the rtc state event http requests.

* use then catch for startup

* no try catch 1

* update expire headroom logic
transition from try catch to .then .catch

* replace flushPromise with advanceTimersByTimeAsync

* fix leaving special cases

* more unrecoverable errors special cases

* move to MatrixRTCSessionManager logger

* add state reset and add another unhandleable error
The error occurs if we want to cancel the delayed event we still have an id for but get a non expected error.

* missed review fixes

* remove @jest/environment dependency

* Cleanup awaits and Make mock types more correct.
Make every mock return a Promise if the real implementation does return a pormise.

* remove flush promise dependency

* fix not recreating default state on reset
This broke all tests since we only created the state once and than passed by ref

* Use per action rate limit and retry counter
There can be multiple retries at once so we need to store counters per action
e.g. the send update membership and the restart delayed could be rate limited at the same time.

* add linting to matrixrtc tests

* Add fix async lints and use matrix rtc logger for test environment.

* prettier

* review step 1

* change to MatrixRTCSession logger

* review step 2

* make LoopHandler Private

* update config to use NewManager wording

* emit error on rtc session if the membership manager encounters one

* network error and throw refactor

* make accessing the full room deprecated

* remove deprecated usage of full room

* Clean up the deprecation

* add network error handler and cleanup

* better logging, another test, make maximumNetworkErrorRetryCount configurable

* more logging & refactor leave promise

* add ConnectionError as possible retry cause

* Make it work in embedded mode with a server that does not support delayed events

* review iteration 1

* review iteration 2

* first step in improving widget error handling

* make the embedded client throw ConnectionErrors where desired.

* fix tests

* delayed event sending widget mode stop gap fix.

* improve comment

* fix unrecoverable error joinState (and add JoinStateChanged) emission.

* check that we do not add multipe sendFirstDelayed Events

* also check insertions queue

* always log "Missing own membership: force re-join"

* Do not update the membership if we are in any (a later) state of sending our own state.
The scheduled states MembershipActionType.SendFirstDelayedEvent and MembershipActionType.SendJoinEvent both imply that we are already trying to send our own membership state event.

* make leave reset actually stop the manager.
The reset case was not covered properly. There are cases where it is not allowed to add additional events after a reset and cases where we want to add more events after the reset. We need to allow this as a reset property.

* fix tests (and implementation)

* Allow MembershipManger to be set at runtime via JoinConfig.membershipManagerFactory

* Map actions into status as a sanity check

* Log status change after applying actions

* Add todo

* Cleanup

* Log transition from earlier status

* remove redundant status implementation
also add TODO comment to not forget about this.

* More cleanup

* Consider insertions in status()

* Log duration for emitting MatrixRTCSessionEvent.MembershipsChanged

* add another valid condition for connected

* some TODO cleanup

* review add warning when using addAction while the scheduler is not running.

* es lint

* refactor to return based handler approach (remove insertions array)

* refactor: Move action scheduler

* refactor: move different handler cases into separate functions

* linter

* review: delayed events endpoint error

* review

* Suggestions from pair review

* resetState is actually only used internally

* Revert "resetState is actually only used internally"

This reverts commit 6af4730.

* refactor: running is part of the scheduler (not state)

* refactor: move everything state related from schduler to manager.

* review

* Update src/matrixrtc/NewMembershipManager.ts

Co-authored-by: Hugh Nimmo-Smith <[email protected]>

* review

* public -> private + missed review fiexes (comment typos)

---------

Co-authored-by: Hugh Nimmo-Smith <[email protected]>
Co-authored-by: Hugh Nimmo-Smith <[email protected]>

Author: toger5

Diff for: spec/unit/embedded.spec.ts

@@ -49,26 +49,26 @@ const testOIDCToken = {
     token_type: "Bearer",
 };
 class MockWidgetApi extends EventEmitter {
-    public start = jest.fn();
-    public requestCapability = jest.fn();
-    public requestCapabilities = jest.fn();
-    public requestCapabilityForRoomTimeline = jest.fn();
-    public requestCapabilityToSendEvent = jest.fn();
-    public requestCapabilityToReceiveEvent = jest.fn();
-    public requestCapabilityToSendMessage = jest.fn();
-    public requestCapabilityToReceiveMessage = jest.fn();
-    public requestCapabilityToSendState = jest.fn();
-    public requestCapabilityToReceiveState = jest.fn();
-    public requestCapabilityToSendToDevice = jest.fn();
-    public requestCapabilityToReceiveToDevice = jest.fn();
+    public start = jest.fn().mockResolvedValue(undefined);
+    public requestCapability = jest.fn().mockResolvedValue(undefined);
+    public requestCapabilities = jest.fn().mockResolvedValue(undefined);
+    public requestCapabilityForRoomTimeline = jest.fn().mockResolvedValue(undefined);
+    public requestCapabilityToSendEvent = jest.fn().mockResolvedValue(undefined);
+    public requestCapabilityToReceiveEvent = jest.fn().mockResolvedValue(undefined);
+    public requestCapabilityToSendMessage = jest.fn().mockResolvedValue(undefined);
+    public requestCapabilityToReceiveMessage = jest.fn().mockResolvedValue(undefined);
+    public requestCapabilityToSendState = jest.fn().mockResolvedValue(undefined);
+    public requestCapabilityToReceiveState = jest.fn().mockResolvedValue(undefined);
+    public requestCapabilityToSendToDevice = jest.fn().mockResolvedValue(undefined);
+    public requestCapabilityToReceiveToDevice = jest.fn().mockResolvedValue(undefined);
     public sendRoomEvent = jest.fn(
-        (eventType: string, content: unknown, roomId?: string, delay?: number, parentDelayId?: string) =>
+        async (eventType: string, content: unknown, roomId?: string, delay?: number, parentDelayId?: string) =>
             delay === undefined && parentDelayId === undefined
                 ? { event_id: `$${Math.random()}` }
                 : { delay_id: `id-${Math.random()}` },
     );
     public sendStateEvent = jest.fn(
-        (
+        async (
             eventType: string,
             stateKey: string,
             content: unknown,
@@ -80,17 +80,17 @@ class MockWidgetApi extends EventEmitter {
                 ? { event_id: `$${Math.random()}` }
                 : { delay_id: `id-${Math.random()}` },
     );
-    public updateDelayedEvent = jest.fn();
-    public sendToDevice = jest.fn();
-    public requestOpenIDConnectToken = jest.fn(() => {
+    public updateDelayedEvent = jest.fn().mockResolvedValue(undefined);
+    public sendToDevice = jest.fn().mockResolvedValue(undefined);
+    public requestOpenIDConnectToken = jest.fn(async () => {
         return testOIDCToken;
         return new Promise<IOpenIDCredentials>(() => {
             return testOIDCToken;
         });
     });
-    public readStateEvents = jest.fn(() => []);
-    public getTurnServers = jest.fn(() => []);
-    public sendContentLoaded = jest.fn();
+    public readStateEvents = jest.fn(async () => []);
+    public getTurnServers = jest.fn(async () => []);
+    public sendContentLoaded = jest.fn().mockResolvedValue(undefined);
 
     public transport = {
         reply: jest.fn(),

Diff for: spec/unit/matrixrtc/MembershipManager.spec.ts

@@ -19,10 +19,11 @@ limitations under the License.
 
 import { type MockedFunction, type Mock } from "jest-mock";
 
-import { EventType, HTTPError, MatrixError, type Room } from "../../../src";
+import { EventType, HTTPError, MatrixError, UnsupportedDelayedEventsEndpointError, type Room } from "../../../src";
 import { type Focus, type LivekitFocusActive, type SessionMembershipData } from "../../../src/matrixrtc";
-import { LegacyMembershipManager } from "../../../src/matrixrtc/MembershipManager";
+import { LegacyMembershipManager } from "../../../src/matrixrtc/LegacyMembershipManager";
 import { makeMockClient, makeMockRoom, membershipTemplate, mockCallMembership, type MockClient } from "./mocks";
+import { MembershipManager } from "../../../src/matrixrtc/NewMembershipManager";
 import { defer } from "../../../src/utils";
 
 function waitForMockCall(method: MockedFunction<any>, returnVal?: Promise<any>) {
@@ -44,9 +45,10 @@ function createAsyncHandle(method: MockedFunction<any>) {
  * Tests different MembershipManager implementations. Some tests don't apply to `LegacyMembershipManager`
  * use !FailsForLegacy to skip those. See: testEnvironment for more details.
  */
+
 describe.each([
     { TestMembershipManager: LegacyMembershipManager, description: "LegacyMembershipManager" },
-    // { TestMembershipManager: MembershipManager, description: "MembershipManager" },
+    { TestMembershipManager: MembershipManager, description: "MembershipManager" },
 ])("$description", ({ TestMembershipManager }) => {
     let client: MockClient;
     let room: Room;
@@ -244,7 +246,12 @@ describe.each([
                 const delayedHandle = createAsyncHandle(client._unstable_sendDelayedStateEvent as Mock);
                 const manager = new TestMembershipManager({}, room, client, () => undefined);
                 manager.join([focus], focusActive);
-                delayedHandle.reject?.(Error("Server does not support the delayed events API"));
+                delayedHandle.reject?.(
+                    new UnsupportedDelayedEventsEndpointError(
+                        "Server does not support the delayed events API",
+                        "sendDelayedStateEvent",
+                    ),
+                );
                 expect(client._unstable_sendDelayedStateEvent).toHaveBeenCalledTimes(1);
             });
             it("does try to schedule a delayed leave event again if rate limited", async () => {
@@ -328,6 +335,7 @@ describe.each([
             await jest.advanceTimersByTimeAsync(1);
             (client._unstable_updateDelayedEvent as Mock<any>).mockRejectedValue("unknown");
             await manager.leave();
+
             // We send a normal leave event since we failed using updateDelayedEvent with the "send" action.
             expect(client.sendStateEvent).toHaveBeenLastCalledWith(
                 room.roomId,
@@ -337,9 +345,9 @@ describe.each([
             );
         });
         // FailsForLegacy because legacy implementation always sends the empty state event even though it isn't needed
-        it("does nothing if not joined !FailsForLegacy", async () => {
+        it("does nothing if not joined !FailsForLegacy", () => {
             const manager = new TestMembershipManager({}, room, client, () => undefined);
-            await manager.leave();
+            expect(async () => await manager.leave()).not.toThrow();
             expect(client._unstable_sendDelayedStateEvent).not.toHaveBeenCalled();
             expect(client.sendStateEvent).not.toHaveBeenCalled();
         });
@@ -470,10 +478,10 @@ describe.each([
         // !FailsForLegacy because the expires logic was removed for the legacy call manager.
         // Delayed events should replace it entirely but before they have wide adoption
         // the expiration logic still makes sense.
-        // TODO: add git commit when we removed it.
-        it("extends `expires` when call still active !FailsForLegacy", async () => {
+        // TODO: Add git commit when we removed it.
+        async function testExpires(expire: number, headroom?: number) {
             const manager = new TestMembershipManager(
-                { membershipExpiryTimeout: 10_000 },
+                { membershipExpiryTimeout: expire, membershipExpiryTimeoutHeadroom: headroom },
                 room,
                 client,
                 () => undefined,
@@ -482,13 +490,19 @@ describe.each([
             await waitForMockCall(client.sendStateEvent);
             expect(client.sendStateEvent).toHaveBeenCalledTimes(1);
             const sentMembership = (client.sendStateEvent as Mock).mock.calls[0][2] as SessionMembershipData;
-            expect(sentMembership.expires).toBe(10_000);
+            expect(sentMembership.expires).toBe(expire);
             for (let i = 2; i <= 12; i++) {
-                await jest.advanceTimersByTimeAsync(10_000);
+                await jest.advanceTimersByTimeAsync(expire);
                 expect(client.sendStateEvent).toHaveBeenCalledTimes(i);
                 const sentMembership = (client.sendStateEvent as Mock).mock.lastCall![2] as SessionMembershipData;
-                expect(sentMembership.expires).toBe(10_000 * i);
+                expect(sentMembership.expires).toBe(expire * i);
             }
+        }
+        it("extends `expires` when call still active !FailsForLegacy", async () => {
+            await testExpires(10_000);
+        });
+        it("extends `expires` using headroom configuration !FailsForLegacy", async () => {
+            await testExpires(10_000, 1_000);
         });
     });
 
@@ -544,7 +558,7 @@ describe.each([
                 expect(client.sendStateEvent).toHaveBeenCalledTimes(1);
             });
             // FailsForLegacy as implementation does not re-check membership before retrying.
-            it("abandons retry loop if leave() was called !FailsForLegacy", async () => {
+            it("abandons retry loop if leave() was called before sending state event !FailsForLegacy", async () => {
                 const handle = createAsyncHandle(client._unstable_sendDelayedStateEvent);
 
                 const manager = new TestMembershipManager({}, room, client, () => undefined);
@@ -565,7 +579,6 @@ describe.each([
                 await manager.leave();
 
                 // Wait for all timers to be setup
-                // await flushPromises();
                 await jest.advanceTimersByTimeAsync(1000);
 
                 // No new events should have been sent:
@@ -603,4 +616,87 @@ describe.each([
             });
         });
     });
+    describe("unrecoverable errors", () => {
+        // !FailsForLegacy because legacy does not have a retry limit and no mechanism to communicate unrecoverable errors.
+        it("throws, when reaching maximum number of retries for initial delayed event creation !FailsForLegacy", async () => {
+            const delayEventSendError = jest.fn();
+            (client._unstable_sendDelayedStateEvent as Mock<any>).mockRejectedValue(
+                new MatrixError(
+                    { errcode: "M_LIMIT_EXCEEDED" },
+                    429,
+                    undefined,
+                    undefined,
+                    new Headers({ "Retry-After": "2" }),
+                ),
+            );
+            const manager = new TestMembershipManager({}, room, client, () => undefined);
+            manager.join([focus], focusActive, delayEventSendError);
+
+            for (let i = 0; i < 10; i++) {
+                await jest.advanceTimersByTimeAsync(2000);
+            }
+            expect(delayEventSendError).toHaveBeenCalled();
+        });
+        // !FailsForLegacy because legacy does not have a retry limit and no mechanism to communicate unrecoverable errors.
+        it("throws, when reaching maximum number of retries !FailsForLegacy", async () => {
+            const delayEventRestartError = jest.fn();
+            (client._unstable_updateDelayedEvent as Mock<any>).mockRejectedValue(
+                new MatrixError(
+                    { errcode: "M_LIMIT_EXCEEDED" },
+                    429,
+                    undefined,
+                    undefined,
+                    new Headers({ "Retry-After": "1" }),
+                ),
+            );
+            const manager = new TestMembershipManager({}, room, client, () => undefined);
+            manager.join([focus], focusActive, delayEventRestartError);
+
+            for (let i = 0; i < 10; i++) {
+                await jest.advanceTimersByTimeAsync(1000);
+            }
+            expect(delayEventRestartError).toHaveBeenCalled();
+        });
+        it("falls back to using pure state events when some error occurs while sending delayed events !FailsForLegacy", async () => {
+            const unrecoverableError = jest.fn();
+            (client._unstable_sendDelayedStateEvent as Mock<any>).mockRejectedValue(new HTTPError("unknown", 601));
+            const manager = new TestMembershipManager({}, room, client, () => undefined);
+            manager.join([focus], focusActive, unrecoverableError);
+            await waitForMockCall(client.sendStateEvent);
+            expect(unrecoverableError).not.toHaveBeenCalledWith();
+            expect(client.sendStateEvent).toHaveBeenCalled();
+        });
+        it("retries before failing in case its a network error !FailsForLegacy", async () => {
+            const unrecoverableError = jest.fn();
+            (client._unstable_sendDelayedStateEvent as Mock<any>).mockRejectedValue(new HTTPError("unknown", 501));
+            const manager = new TestMembershipManager(
+                { callMemberEventRetryDelayMinimum: 1000, maximumNetworkErrorRetryCount: 7 },
+                room,
+                client,
+                () => undefined,
+            );
+            manager.join([focus], focusActive, unrecoverableError);
+            for (let retries = 0; retries < 7; retries++) {
+                expect(client._unstable_sendDelayedStateEvent).toHaveBeenCalledTimes(retries + 1);
+                await jest.advanceTimersByTimeAsync(1000);
+            }
+            expect(unrecoverableError).toHaveBeenCalled();
+            expect(unrecoverableError.mock.lastCall![0].message).toMatch(
+                "The MembershipManager shut down because of the end condition",
+            );
+            expect(client.sendStateEvent).not.toHaveBeenCalled();
+        });
+        it("falls back to using pure state events when UnsupportedDelayedEventsEndpointError encountered for delayed events !FailsForLegacy", async () => {
+            const unrecoverableError = jest.fn();
+            (client._unstable_sendDelayedStateEvent as Mock<any>).mockRejectedValue(
+                new UnsupportedDelayedEventsEndpointError("not supported", "sendDelayedStateEvent"),
+            );
+            const manager = new TestMembershipManager({}, room, client, () => undefined);
+            manager.join([focus], focusActive, unrecoverableError);
+            await jest.advanceTimersByTimeAsync(1);
+
+            expect(unrecoverableError).not.toHaveBeenCalled();
+            expect(client.sendStateEvent).toHaveBeenCalled();
+        });
+    });
 });

Diff for: spec/unit/matrixrtc/mocks.ts

@@ -25,7 +25,7 @@ export const membershipTemplate: SessionMembershipData = {
     call_id: "",
     device_id: "AAAAAAA",
     scope: "m.room",
-    focus_active: { type: "livekit", livekit_service_url: "https://lk.url" },
+    focus_active: { type: "livekit", focus_selection: "oldest_membership" },
     foci_preferred: [
         {
             livekit_alias: "!alias:something.org",

Diff for: src/client.ts

@@ -240,6 +240,7 @@ import {
     validateAuthMetadataAndKeys,
 } from "./oidc/index.ts";
 import { type EmptyObject } from "./@types/common.ts";
+import { UnsupportedDelayedEventsEndpointError } from "./errors.ts";
 
 export type Store = IStore;
 
@@ -3351,7 +3352,10 @@ export class MatrixClient extends TypedEventEmitter<EmittedEvents, ClientEventHa
         txnId?: string,
     ): Promise<SendDelayedEventResponse> {
         if (!(await this.doesServerSupportUnstableFeature(UNSTABLE_MSC4140_DELAYED_EVENTS))) {
-            throw Error("Server does not support the delayed events API");
+            throw new UnsupportedDelayedEventsEndpointError(
+                "Server does not support the delayed events API",
+                "sendDelayedEvent",
+            );
         }
 
         this.addThreadRelationIfNeeded(content, threadId, roomId);
@@ -3374,7 +3378,10 @@ export class MatrixClient extends TypedEventEmitter<EmittedEvents, ClientEventHa
         opts: IRequestOpts = {},
     ): Promise<SendDelayedEventResponse> {
         if (!(await this.doesServerSupportUnstableFeature(UNSTABLE_MSC4140_DELAYED_EVENTS))) {
-            throw Error("Server does not support the delayed events API");
+            throw new UnsupportedDelayedEventsEndpointError(
+                "Server does not support the delayed events API",
+                "sendDelayedStateEvent",
+            );
         }
 
         const pathParams = {
@@ -3398,7 +3405,10 @@ export class MatrixClient extends TypedEventEmitter<EmittedEvents, ClientEventHa
     // eslint-disable-next-line
     public async _unstable_getDelayedEvents(fromToken?: string): Promise<DelayedEventInfo> {
         if (!(await this.doesServerSupportUnstableFeature(UNSTABLE_MSC4140_DELAYED_EVENTS))) {
-            throw Error("Server does not support the delayed events API");
+            throw new UnsupportedDelayedEventsEndpointError(
+                "Server does not support the delayed events API",
+                "getDelayedEvents",
+            );
         }
 
         const queryDict = fromToken ? { from: fromToken } : undefined;
@@ -3420,7 +3430,10 @@ export class MatrixClient extends TypedEventEmitter<EmittedEvents, ClientEventHa
         requestOptions: IRequestOpts = {},
     ): Promise<EmptyObject> {
         if (!(await this.doesServerSupportUnstableFeature(UNSTABLE_MSC4140_DELAYED_EVENTS))) {
-            throw Error("Server does not support the delayed events API");
+            throw new UnsupportedDelayedEventsEndpointError(
+                "Server does not support the delayed events API",
+                "updateDelayedEvent",
+            );
         }
 
         const path = utils.encodeUri("/delayed_events/$delayId", {