From 6a9583a2ce2738c6b1aa95e29e47e35662c97685 Mon Sep 17 00:00:00 2001 From: Robin Date: Mon, 7 Sep 2026 12:53:47 +0200 Subject: [PATCH 1/6] Adapt delayed leave timings when delegation is available Splits the config options for the timings of a delayed leave event into two sets: one for when delegation is available (as you can relax the timings and get more stable calls this way), and another for when it's unavailable (as we must continue to gracefully downgrade even after Matrix 2.0 is fully rolled out). This works by bluntly hitting the delegation endpoints without auth before joining to check for a 404. --- config/config.devenv.json | 2 - config/config.sample.json | 2 - config/config_netlify_preview.json | 2 - config/config_netlify_preview_sdk.json | 2 - src/config/ConfigOptions.ts | 88 +++--- src/livekit/openIDSFU.test.ts | 8 +- src/livekit/openIDSFU.ts | 10 +- src/state/CallViewModel/CallViewModel.ts | 29 +- .../localMember/LocalMember.test.ts | 270 +++++++++++------- .../CallViewModel/localMember/LocalMember.ts | 123 +++++--- src/utils/observable.test.ts | 23 +- src/utils/observable.ts | 10 +- vite-embedded.config.ts | 2 - 13 files changed, 363 insertions(+), 208 deletions(-) diff --git a/config/config.devenv.json b/config/config.devenv.json index 48602406b..294209a67 100644 --- a/config/config.devenv.json +++ b/config/config.devenv.json @@ -9,8 +9,6 @@ "matrix_rtc_session": { "wait_for_key_rotation_ms": 3000, "membership_event_expiry_ms": 180000000, - "delayed_leave_event_delay_ms": 18000, - "delayed_leave_event_restart_ms": 4000, "network_error_retry_ms": 100 } } diff --git a/config/config.sample.json b/config/config.sample.json index 0baa522e4..9f7ab61ac 100644 --- a/config/config.sample.json +++ b/config/config.sample.json @@ -13,8 +13,6 @@ "matrix_rtc_session": { "wait_for_key_rotation_ms": 3000, "membership_event_expiry_ms": 180000000, - "delayed_leave_event_delay_ms": 18000, - "delayed_leave_event_restart_ms": 4000, "network_error_retry_ms": 100 } } diff --git a/config/config_netlify_preview.json b/config/config_netlify_preview.json index 313f0d02c..179177bac 100644 --- a/config/config_netlify_preview.json +++ b/config/config_netlify_preview.json @@ -9,8 +9,6 @@ "matrix_rtc_session": { "wait_for_key_rotation_ms": 3000, "membership_event_expiry_ms": 180000000, - "delayed_leave_event_delay_ms": 18000, - "delayed_leave_event_restart_ms": 4000, "network_error_retry_ms": 100 }, "posthog": { diff --git a/config/config_netlify_preview_sdk.json b/config/config_netlify_preview_sdk.json index 784f0c7ef..fa60fe8d4 100644 --- a/config/config_netlify_preview_sdk.json +++ b/config/config_netlify_preview_sdk.json @@ -9,8 +9,6 @@ "matrix_rtc_session": { "wait_for_key_rotation_ms": 3000, "membership_event_expiry_ms": 180000000, - "delayed_leave_event_delay_ms": 18000, - "delayed_leave_event_restart_ms": 4000, "network_error_retry_ms": 100 } } diff --git a/src/config/ConfigOptions.ts b/src/config/ConfigOptions.ts index 536966e00..872e35168 100644 --- a/src/config/ConfigOptions.ts +++ b/src/config/ConfigOptions.ts @@ -24,6 +24,35 @@ export enum MatrixRTCMode { Matrix_2_0 = "matrix_2_0", } +export interface DelayedLeaveTimings { + /** + * The delay (in milliseconds) with which delayed leave events are sent. + * + * If the server receives no keep-alives from the client for any longer than + * this duration, it will send the leave event, automatically removing the + * user from the call. + */ + delay_ms?: number; + + /** + * How frequently (in milliseconds) the client sends keep-alives to the server + * to restart the timer for a delayed leave event. Should be less than + * {@link DelayedLeaveTimings.delay_ms}. + */ + restart_ms?: number; + + /** + * The time (in milliseconds) after which we consider a delayed event restart HTTP request to have failed. + * Setting this to a lower value will result in more frequent retries, but then we will also give up earlier. + * + * In the presence of network packet loss (hurting TCP connections), the custom delayedEventRestartLocalTimeoutMs + * helps by keeping more delayed event reset candidates in flight, + * improving the chances of a successful reset. (its is equivalent to the js-sdk `localTimeout` configuration, + * but only applies to calls to the `_unstable_updateDelayedEvent` endpoint with a body of `{action:"restart"}`.) + */ + restart_timeout_ms?: number; +} + export interface ConfigOptions { /** * The Posthog endpoint to which analytics data will be sent. @@ -184,29 +213,6 @@ export interface ConfigOptions { */ wait_for_key_rotation_ms?: number; - /** - * The duration (in milliseconds) after the most recent keep-alive (delayed leave event restart) - * that the server waits before sending the leave MatrixRTC membership event. - */ - delayed_leave_event_delay_ms?: number; - - /** - * The time (in milliseconds) after which we consider a delayed event restart http request to have failed. - * Setting this to a lower value will result in more frequent retries but also a higher chance of failiour. - * - * In the presence of network packet loss (hurting TCP connections), the custom delayedEventRestartLocalTimeoutMs - * helps by keeping more delayed event reset candidates in flight, - * improving the chances of a successful reset. (its is equivalent to the js-sdk `localTimeout` configuration, - * but only applies to calls to the `_unstable_updateDelayedEvent` endpoint with a body of `{action:"restart"}`.) - */ - delayed_leave_event_restart_local_timeout_ms?: number; - - /** - * The time interval (in milliseconds) at which the client sends membership keep-alive - * messages to the server by restarting the timer for the delayed leave event. - */ - delayed_leave_event_restart_ms?: number; - /** * How long we wait before retrying after a network error on any of the requests. */ @@ -231,9 +237,28 @@ export interface ConfigOptions { * Defaults to the js-sdk default (undefined). Which means that rotation will always happen. */ key_rotation_participant_limit?: number; + + /** + * Timing options for delayed leave events, which are used to remove a user + * from a call when they lose connection. + */ + delayed_leave?: DelayedLeaveTimings; + + /** + * Timing options for delayed leave events, in cases where the ability to + * send the event can be delegated to the SFU. + * + * We recommend setting {@link DelayedLeaveTimings.delay_ms} >> + * {@link sync_disconnect_grace_period_ms} here. + */ + delegated_delayed_leave?: DelayedLeaveTimings; }; } +export interface ResolvedDelayedLeaveTimings extends DelayedLeaveTimings { + delay_ms: number; // Required +} + // Overrides members from ConfigOptions that are always provided by the // default config and are therefore non-optional. export interface ResolvedConfigOptions extends ConfigOptions { @@ -257,19 +282,15 @@ export interface ResolvedConfigOptions extends ConfigOptions { > >; }; - matrix_rtc_session: { - wait_for_key_rotation_ms?: number; - delayed_leave_event_delay_ms: number; - delayed_leave_event_restart_local_timeout_ms?: number; - delayed_leave_event_restart_ms?: number; + matrix_rtc_session: ConfigOptions["matrix_rtc_session"] & { network_error_retry_ms: number; - membership_event_expiry_ms?: number; - key_rotation_participant_limit?: number; + delayed_leave: ResolvedDelayedLeaveTimings; + delegated_delayed_leave: ResolvedDelayedLeaveTimings; }; } export const DEFAULT_CONFIG: ResolvedConfigOptions = { - sync_disconnect_grace_period_ms: 10000, + sync_disconnect_grace_period_ms: 10_000, ssla: "https://static.element.io/legal/element-software-and-services-license-agreement-uk-1.pdf", media_quality: { video_codec: "vp8", @@ -285,7 +306,8 @@ export const DEFAULT_CONFIG: ResolvedConfigOptions = { }, }, matrix_rtc_session: { - delayed_leave_event_delay_ms: 10000, - network_error_retry_ms: 1000, + network_error_retry_ms: 1_000, + delayed_leave: { delay_ms: 18_000, restart_ms: 4_000 }, + delegated_delayed_leave: { delay_ms: 3_600_000, restart_ms: 300_000 }, }, }; diff --git a/src/livekit/openIDSFU.test.ts b/src/livekit/openIDSFU.test.ts index 2ddb6c95c..f759a3aac 100644 --- a/src/livekit/openIDSFU.test.ts +++ b/src/livekit/openIDSFU.test.ts @@ -148,7 +148,7 @@ describe("getSFUConfigWithOpenID", () => { // Verify, that the request contains the expected delay parameters if ( body.delay_id === "mock_delay_id" && - body.delay_timeout === 10000 && + body.delay_timeout === 3600000 && body.delay_cs_api_url === "https://homeserverserver.org/cs_api" ) { return { @@ -229,7 +229,7 @@ describe("getSFUConfigWithOpenID", () => { expect(calls[0][0]).toStrictEqual("https://sfu.example.org/get_token"); expect(calls[0][1]).toStrictEqual({ // check if it uses correct delayID! - body: '{"room_id":"!example_room_id","slot_id":"m.call#ROOM","member":{"id":"@alice:example.org:DEVICE","claimed_user_id":"@alice:example.org","claimed_device_id":"DEVICE"},"delay_id":"mock_delay_id","delay_timeout":10000,"delay_cs_api_url":"https://matrix.homeserverserver.org"}', + body: '{"room_id":"!example_room_id","slot_id":"m.call#ROOM","member":{"id":"@alice:example.org:DEVICE","claimed_user_id":"@alice:example.org","claimed_device_id":"DEVICE"},"delay_id":"mock_delay_id","delay_timeout":3600000,"delay_cs_api_url":"https://matrix.homeserverserver.org"}', method: "POST", headers: { "Content-Type": "application/json", @@ -239,7 +239,7 @@ describe("getSFUConfigWithOpenID", () => { expect(calls[1][0]).toStrictEqual("https://sfu.example.org/sfu/get"); expect(calls[1][1]).toStrictEqual({ - body: '{"room":"!example_room_id","device_id":"DEVICE","delay_id":"mock_delay_id","delay_timeout":10000,"delay_cs_api_url":"https://matrix.homeserverserver.org"}', + body: '{"room":"!example_room_id","device_id":"DEVICE","delay_id":"mock_delay_id","delay_timeout":3600000,"delay_cs_api_url":"https://matrix.homeserverserver.org"}', headers: { "Content-Type": "application/json", }, @@ -284,7 +284,7 @@ describe("getSFUConfigWithOpenID", () => { expect(calls[0][0]).toStrictEqual("https://sfu.example.org/get_token"); expect(calls[0][1]).toStrictEqual({ // check if it uses correct delayID! - body: '{"room_id":"!example_room_id","slot_id":"m.call#ROOM","member":{"id":"@alice:example.org:DEVICE","claimed_user_id":"@alice:example.org","claimed_device_id":"DEVICE"},"delay_id":"mock_delay_id","delay_timeout":10000,"delay_cs_api_url":"https://matrix.homeserverserver.org"}', + body: '{"room_id":"!example_room_id","slot_id":"m.call#ROOM","member":{"id":"@alice:example.org:DEVICE","claimed_user_id":"@alice:example.org","claimed_device_id":"DEVICE"},"delay_id":"mock_delay_id","delay_timeout":3600000,"delay_cs_api_url":"https://matrix.homeserverserver.org"}', method: "POST", headers: { "Content-Type": "application/json", diff --git a/src/livekit/openIDSFU.ts b/src/livekit/openIDSFU.ts index 00cf69b1a..67dd5e9e7 100644 --- a/src/livekit/openIDSFU.ts +++ b/src/livekit/openIDSFU.ts @@ -204,11 +204,10 @@ async function getLiveKitJWT( let bodyDalayParts: IDelayParams = {}; // Also check for empty string if (delayId && delayEndpointBaseUrl) { - const delayTimeoutMs = - Config.get().matrix_rtc_session?.delayed_leave_event_delay_ms; bodyDalayParts = { delay_id: delayId, - delay_timeout: delayTimeoutMs, + delay_timeout: + Config.get().matrix_rtc_session.delegated_delayed_leave.delay_ms, delay_cs_api_url: delayEndpointBaseUrl, }; } @@ -288,11 +287,10 @@ export async function getLiveKitJWTWithDelayDelegation( let bodyDalayParts = {}; // Also check for empty string if (delayId && delayEndpointBaseUrl) { - const delayTimeoutMs = - Config.get().matrix_rtc_session?.delayed_leave_event_delay_ms; bodyDalayParts = { delay_id: delayId, - delay_timeout: delayTimeoutMs, + delay_timeout: + Config.get().matrix_rtc_session.delegated_delayed_leave.delay_ms, delay_cs_api_url: delayEndpointBaseUrl, }; } diff --git a/src/state/CallViewModel/CallViewModel.ts b/src/state/CallViewModel/CallViewModel.ts index c854363be..62f05a476 100644 --- a/src/state/CallViewModel/CallViewModel.ts +++ b/src/state/CallViewModel/CallViewModel.ts @@ -64,7 +64,10 @@ import { showReactions, } from "../../settings/settings"; import { Config } from "../../config/Config"; -import { MatrixRTCMode } from "../../config/ConfigOptions"; +import { + MatrixRTCMode, + type ResolvedDelayedLeaveTimings, +} from "../../config/ConfigOptions"; import { isFirefox, platform } from "../../Platform"; import { setPipEnabled$ } from "../../controls"; import { TileStore } from "../TileStore"; @@ -562,16 +565,6 @@ export function createCallViewModel$( localUser: { userId, deviceId }, }); - const connectOptions$ = scope.behavior( - matrixRTCMode$.pipe( - map((mode) => ({ - encryptMedia: livekitKeyProvider !== undefined, - // TODO. This might need to get called again on each change of matrixRTCMode... - matrixRTCMode: mode, - })), - ), - ); - const localMembership = createLocalMembership$({ scope, homeserverConnected: createHomeserverConnected$( @@ -580,12 +573,21 @@ export function createCallViewModel$( matrixRTCSession, ), muteStates, - joinMatrixRTC: (transport: LivekitTransportConfig) => { + joinMatrixRTC: ( + transport: LivekitTransportConfig, + delayedLeaveTimings: ResolvedDelayedLeaveTimings, + ) => { return enterRTCSession( matrixRTCSession, ownMembershipIdentity, transport, - connectOptions$.value, + { + encryptMedia: livekitKeyProvider !== undefined, + // We merely sample the current mode here, so the user would need to + // manually rejoin to switch to a different one + matrixRTCMode: matrixRTCMode$.value, + delayedLeaveTimings, + }, ); }, createPublisherFactory: (connection: Connection) => { @@ -603,6 +605,7 @@ export function createCallViewModel$( matrixRTCSession, localTransport$, roomId: matrixRoom.roomId, + baseUrl: client.baseUrl, logger: logger.getChild(`[${Date.now()}]`), }); diff --git a/src/state/CallViewModel/localMember/LocalMember.test.ts b/src/state/CallViewModel/localMember/LocalMember.test.ts index aec641776..66cd8cda0 100644 --- a/src/state/CallViewModel/localMember/LocalMember.test.ts +++ b/src/state/CallViewModel/localMember/LocalMember.test.ts @@ -19,13 +19,18 @@ import { beforeAll, afterAll, beforeEach, + afterEach, } from "vitest"; import { BehaviorSubject, map, of } from "rxjs"; import { logger } from "matrix-js-sdk/lib/logger"; import { type LocalParticipant, type LocalTrack } from "livekit-client"; +import fetchMock from "fetch-mock"; import { PosthogAnalytics } from "../../../analytics/PosthogAnalytics"; -import { MatrixRTCMode } from "../../../config/ConfigOptions"; +import { + MatrixRTCMode, + type ResolvedDelayedLeaveTimings, +} from "../../../config/ConfigOptions"; import { type HomeserverDisconnectReason } from "./HomeserverConnected"; import { flushPromises, @@ -35,6 +40,7 @@ import { mockMuteStates, withTestScheduler, ownMemberMock, + testScope, } from "../../../utils/test"; import { TransportState, @@ -95,112 +101,109 @@ describe("watchScreenShareToggle", () => { }); }); -describe("LocalMembership", () => { - describe("enterRTCSession", () => { - it("It joins the correct Session", () => { - mockConfig({ - livekit: { livekit_service_url: "http://my-default-service-url.com" }, - }); +const timings: ResolvedDelayedLeaveTimings = { + delay_ms: 10000, + restart_ms: 4000, + restart_timeout_ms: 1000, +}; - const mockedSession = vi.mocked({ - room: { - roomId: "roomId", - client: { - getDomain: vi.fn().mockReturnValue("example.org"), - getOpenIdToken: vi.fn().mockResolvedValue({ - access_token: "ACCCESS_TOKEN", - token_type: "Bearer", - matrix_server_name: "localhost", - expires_in: 10000, - }), - }, - }, - memberships: [], - joinRTCSession: vi.fn(), - }) as unknown as MatrixRTCSession; +const delegatedTimings: ResolvedDelayedLeaveTimings = { + delay_ms: timings.delay_ms * 10, + restart_ms: timings.restart_ms! * 10, + restart_timeout_ms: timings.restart_timeout_ms! * 10, +}; - enterRTCSession( - mockedSession, - ownMemberMock, - { - livekit_alias: "roomId", - livekit_service_url: "http://my-livekit-service-url.com", - type: "livekit", - }, - { - encryptMedia: true, - matrixRTCMode: MATRIX_RTC_MODE, - }, - ); +describe("enterRTCSession", () => { + const transport: LivekitTransportConfig = { + livekit_alias: "roomId", + livekit_service_url: "http://my-livekit-service-url.com", + type: "livekit", + }; - expect(mockedSession.joinRTCSession).toHaveBeenLastCalledWith( - { - deviceId: "DEVICE", - memberId: "@alice:example.org:DEVICE", - userId: "@alice:example.org", - }, - [], - { - livekit_alias: "roomId", - livekit_service_url: "http://my-livekit-service-url.com", - type: "livekit", - }, - expect.objectContaining({ manageMediaKeys: true }), - ); - }); + const options = { + encryptMedia: true, + matrixRTCMode: MATRIX_RTC_MODE, + delayedLeaveTimings: timings, + }; - it("passes keyRotationParticipantLimit from config to joinRTCSession", () => { - mockConfig({ - livekit: { livekit_service_url: "http://my-default-service-url.com" }, - matrix_rtc_session: { - delayed_leave_event_delay_ms: 0, - network_error_retry_ms: 0, - key_rotation_participant_limit: 50, - }, - }); - - const mockedSession = vi.mocked({ - room: { - roomId: "roomId", - client: { - getDomain: vi.fn().mockReturnValue("example.org"), - getOpenIdToken: vi.fn().mockResolvedValue({ - access_token: "ACCCESS_TOKEN", - token_type: "Bearer", - matrix_server_name: "localhost", - expires_in: 10000, - }), - }, - }, - memberships: [], - joinRTCSession: vi.fn(), - }) as unknown as MatrixRTCSession; - - enterRTCSession( - mockedSession, - ownMemberMock, - { - livekit_alias: "roomId", - livekit_service_url: "http://my-livekit-service-url.com", - type: "livekit", - }, - { - encryptMedia: true, - matrixRTCMode: MATRIX_RTC_MODE, - }, - ); - - expect(mockedSession.joinRTCSession).toHaveBeenLastCalledWith( - expect.any(Object), - [], - expect.any(Object), - expect.objectContaining({ - keyRotationParticipantLimit: 50, + const mockedSession = vi.mocked({ + room: { + roomId: "roomId", + client: { + getDomain: vi.fn().mockReturnValue("example.org"), + getOpenIdToken: vi.fn().mockResolvedValue({ + access_token: "ACCCESS_TOKEN", + token_type: "Bearer", + matrix_server_name: "localhost", + expires_in: 10000, }), - ); - }); + }, + }, + memberships: [], + joinRTCSession: vi.fn(), + }) as unknown as MatrixRTCSession; + + beforeEach(() => + mockConfig({ + livekit: { livekit_service_url: "http://my-default-service-url.com" }, + }), + ); + + it("It joins the correct Session", () => { + enterRTCSession(mockedSession, ownMemberMock, transport, options); + + expect(mockedSession.joinRTCSession).toHaveBeenLastCalledWith( + { + deviceId: "DEVICE", + memberId: "@alice:example.org:DEVICE", + userId: "@alice:example.org", + }, + [], + transport, + expect.objectContaining({ manageMediaKeys: true }), + ); }); + it("passes keyRotationParticipantLimit from config to joinRTCSession", () => { + mockConfig({ + livekit: { livekit_service_url: "http://my-default-service-url.com" }, + matrix_rtc_session: { + network_error_retry_ms: 0, + key_rotation_participant_limit: 50, + delayed_leave: timings, + delegated_delayed_leave: timings, + }, + }); + + enterRTCSession(mockedSession, ownMemberMock, transport, options); + + expect(mockedSession.joinRTCSession).toHaveBeenLastCalledWith( + expect.any(Object), + [], + expect.any(Object), + expect.objectContaining({ + keyRotationParticipantLimit: 50, + }), + ); + }); + + it("uses the specified delayed leave timings", () => { + enterRTCSession(mockedSession, ownMemberMock, transport, options); + + expect(mockedSession.joinRTCSession).toHaveBeenLastCalledWith( + expect.anything(), + expect.anything(), + expect.anything(), + expect.objectContaining({ + delayedLeaveEventRestartMs: timings.restart_ms, + delayedLeaveEventDelayMs: timings.delay_ms, + delayedLeaveEventRestartLocalTimeoutMs: timings.restart_timeout_ms, + }), + ); + }); +}); + +describe("LocalMembership", () => { const defaultCreateLocalMemberValues = { options: constant({ encryptMedia: false, @@ -226,8 +229,26 @@ describe("LocalMembership", () => { rtsSession$: constant(RTCMemberStatus.Connected), }, roomId: "!test-room-id:example.org", + baseUrl: "https://matrix.example.org", }; + beforeEach(() => { + mockConfig({ + livekit: { livekit_service_url: "http://my-default-service-url.com" }, + matrix_rtc_session: { + network_error_retry_ms: 1000, + delayed_leave: timings, + delegated_delayed_leave: delegatedTimings, + }, + }); + fetchMock.catch(404); + }); + + afterEach(async () => { + void (await fetchMock.flush()); + fetchMock.reset(); + }); + it("throws error on missing RTC config error", () => { withTestScheduler(({ scope, hot, behavior, expectObservable }) => { const localTransport$ = scope.behavior( @@ -256,7 +277,6 @@ describe("LocalMembership", () => { connectionManager: mockConnectionManager, localTransport$: behavior("a", { a: aLocalTransport }), }); - localMembership.requestJoinAndPublish(); expectObservable(localMembership.localMemberState$).toBe("ne", { n: TransportState.Waiting, @@ -299,9 +319,8 @@ describe("LocalMembership", () => { scope, ...defaultCreateLocalMemberValues, connectionManager: mockConnectionManager, - localTransport$: behavior("a", { a: aLocalTransport }), + localTransport$: constant(aLocalTransport), }); - localMembership.requestJoinAndPublish(); expectObservable(localMembership.localMemberState$).toBe("n-e", { n: TransportState.Waiting, @@ -397,6 +416,51 @@ describe("LocalMembership", () => { livekitRoom: mockLivekitRoom({}), } as unknown as Connection; + it.each([ + ["no", null, timings], + [ + "homeserver", + "https://matrix.example.org/_matrix/client/unstable/io.element.msc4195/rtc/livekit/delegate_delayed_leave", + delegatedTimings, + ], + ["transport", "/a/delegate_delayed_leave", delegatedTimings], + ])( + "joins session with %s delegation support", + async (_serviceName, delegationUrl, delayedLeaveTimings) => { + const scope = testScope(); + + const activeTransport$ = constant(aTransportWithSFUConfig); + const aLocalTransport: LocalTransport = { + advertised$: constant(aTransport), + active$: activeTransport$, + }; + const connectionManagerData = new ConnectionManagerData(); + const joinMatrixRTC = vi.fn(); + + if (delegationUrl !== null) + fetchMock.post(delegationUrl, () => ({ status: 401, body: {} })); + + const localMembership = createLocalMembership$({ + scope, + ...defaultCreateLocalMemberValues, + connectionManager: { + connectionManagerData$: constant(new Epoch(connectionManagerData)), + }, + localTransport$: constant(aLocalTransport), + joinMatrixRTC, + }); + + localMembership.requestJoinAndPublish(); + void (await fetchMock.flush()); + await flushPromises(); + // Joins with timings appropriate for the level of delegation support + expect(joinMatrixRTC).toHaveBeenCalledWith( + aTransport, + delayedLeaveTimings, + ); + }, + ); + it("recreates publisher if new connection is used, always unpublish and end tracks", async () => { const scope = new ObservableScope(); diff --git a/src/state/CallViewModel/localMember/LocalMember.ts b/src/state/CallViewModel/localMember/LocalMember.ts index dfe8936a7..838b14ee2 100644 --- a/src/state/CallViewModel/localMember/LocalMember.ts +++ b/src/state/CallViewModel/localMember/LocalMember.ts @@ -62,7 +62,10 @@ import { screenShareCodec, parseResolution, } from "../../../settings/settings.ts"; -import { MatrixRTCMode } from "../../../config/ConfigOptions.ts"; +import { + MatrixRTCMode, + type ResolvedDelayedLeaveTimings, +} from "../../../config/ConfigOptions.ts"; import { Config } from "../../../config/Config.ts"; import { ConnectionState, @@ -72,6 +75,8 @@ import { import { type HomeserverConnected } from "./HomeserverConnected.ts"; import { type LocalTransport } from "./LocalTransport.ts"; import { areLivekitTransportsEqual } from "../remoteMembers/MatrixLivekitMembers.ts"; +import { doNetworkOperationWithRetry } from "../../../utils/matrix.ts"; +import { or$ } from "../../../utils/observable.ts"; export enum TransportState { /** Not even a transport is available to the LocalMembership */ @@ -133,7 +138,10 @@ interface Props { muteStates: MuteStates; connectionManager: IConnectionManager; createPublisherFactory: (connection: Connection) => Publisher; - joinMatrixRTC: (transport: LivekitTransportConfig) => void; + joinMatrixRTC: ( + transport: LivekitTransportConfig, + delayedLeaveTimings: ResolvedDelayedLeaveTimings, + ) => void; homeserverConnected: HomeserverConnected; roomId: string; localTransport$: Behavior; @@ -141,6 +149,7 @@ interface Props { MatrixRTCSession, "updateCallIntent" | "leaveRoomSession" >; + baseUrl: string; logger: Logger; } @@ -159,6 +168,7 @@ interface Props { * @param props.logger The logger to use. * @param props.muteStates The mute states for video and audio. * @param props.matrixRTCSession The matrix RTC session to join. + * @param props.baseUrl Base URL of the homeserver. * @param props.roomId The room ID used as the call identifier in analytics events. * @returns * - publisher: The handle to create tracks and publish them to the room. @@ -177,6 +187,7 @@ export const createLocalMembership$ = ({ logger: parentLogger, muteStates, matrixRTCSession, + baseUrl, roomId, }: Props): { /** @@ -236,11 +247,58 @@ export const createLocalMembership$ = ({ return of(null); }; - // This is the transport that we will advertise in our membership. - const advertisedTransport$ = localTransport$.pipe( + async function checkDelegationSupport( + endpointUrl: string, + serviceName: string, + ): Promise { + logger.info(`Checking whether ${serviceName} supports delegation…`); + try { + // Bluntly hit the endpoint without auth to check for a 404 + const res = await doNetworkOperationWithRetry(async () => + fetch(endpointUrl, { method: "POST" }), + ); + if (res.status === 404) { + logger.warn(`${serviceName} does not support delegation`); + return false; + } else { + logger.info(`${serviceName} supports delegation`); + return true; + } + } catch (e) { + logger.warn( + `Failed to determine whether ${serviceName} supports delegation, assuming no support`, + e, + ); + return false; + } + } + + const homeserverSupportsDelegation = checkDelegationSupport( + baseUrl + + "/_matrix/client/unstable/io.element.msc4195/rtc/livekit/delegate_delayed_leave", + "homeserver", + ); + + // The transport that we will advertise in our membership, paired with info as + // to whether delayed event delegation is supported + const joinParams$ = localTransport$.pipe( switchMap((lt) => lt.advertised$), catchError(handleTransportError), distinctUntilChanged(areLivekitTransportsEqual), + switchMap((transport) => { + if (transport === null) return of(null); + const transportSupportsDelegation = checkDelegationSupport( + transport.livekit_service_url + "/delegate_delayed_leave", + `transport ${transport.livekit_service_url}`, + ); + return or$( + from(homeserverSupportsDelegation), + from(transportSupportsDelegation), + ).pipe( + map((delegationSupported) => ({ transport, delegationSupported })), + startWith(null), + ); + }), ); // Unwrap the local transport and set the state of the LocalMembership to error in case the transport is an error. @@ -617,16 +675,20 @@ export const createLocalMembership$ = ({ // Keep matrix rtc session in sync with advertisedTransport$, connectRequested$ scope.reconcile( - scope.behavior( - combineLatest([advertisedTransport$, joinAndPublishRequested$]), - ), - async ([transport, shouldConnect]) => { - if (!transport) return; + scope.behavior(combineLatest([joinParams$, joinAndPublishRequested$])), + async ([joinParams, shouldConnect]) => { + if (!joinParams) return; // if shouldConnect=false we will do the disconnect as the cleanup from the previous reconcile iteration. if (!shouldConnect) return; + const sessionConfig = Config.get().matrix_rtc_session; try { - joinMatrixRTC(transport); + joinMatrixRTC( + joinParams.transport, + joinParams.delegationSupported + ? sessionConfig.delegated_delayed_leave + : sessionConfig.delayed_leave, + ); } catch (error) { logger.error("Error entering RTC session", error); if (error instanceof Error) @@ -864,6 +926,7 @@ export function observeSharingScreen$(p: Participant): Observable { interface EnterRTCSessionOptions { encryptMedia: boolean; matrixRTCMode: MatrixRTCMode; + delayedLeaveTimings: ResolvedDelayedLeaveTimings; } /** @@ -876,6 +939,7 @@ interface EnterRTCSessionOptions { * @param rtcSession - The MatrixRTCSession to join. * @param ownMembershipIdentity - Options for entering the RTC session. * @param transport - The LivekitTransport to use for this session. + * @param delayedLeaveTimings - The preferred timings for delayed leave events. * @param options - `encryptMedia`: Whether to encrypt media `matrixRTCMode`: The Matrix RTC mode to use. * @throws If the widget could not send ElementWidgetActions.JoinCall action. */ @@ -884,9 +948,8 @@ export function enterRTCSession( rtcSession: MatrixRTCSession, ownMembershipIdentity: CallMembershipIdentityParts, transport: LivekitTransportConfig, - options: EnterRTCSessionOptions, + { encryptMedia, matrixRTCMode, delayedLeaveTimings }: EnterRTCSessionOptions, ): void { - const { encryptMedia, matrixRTCMode } = options; PosthogAnalytics.instance.eventCallEnded.cacheStartCall(new Date()); PosthogAnalytics.instance.eventCallStarted.track(rtcSession.room.roomId); @@ -894,7 +957,11 @@ export function enterRTCSession( // have started tracking by the time calls start getting created. // groupCallOTelMembership?.onJoinCall(); - const { matrix_rtc_session: matrixRtcSessionConfig } = Config.get(); + const { + sync_disconnect_grace_period_ms: gracePeriod, + matrix_rtc_session: sessionConfig, + } = Config.get(); + const retryInterval = sessionConfig.network_error_retry_ms; const { sendNotificationType: notificationType, callIntent } = getUrlParams(); const multiSFU = matrixRTCMode === MatrixRTCMode.Compatibility || @@ -912,16 +979,11 @@ export function enterRTCSession( }; } - // Calculates `maximumNetworkErrorRetryCount`. The connection is failed if EITHER: - // - The /sync loop is unresponsive for > `gracePeriod` ms, or - // - A delayed leave event is emitted (after `leaveDelay` ms period). - // Note: Use leaveDelay >> gracePeriod for delegated leave events. - const gracePeriod = Config.get().sync_disconnect_grace_period_ms; - const leaveDelay = matrixRtcSessionConfig?.delayed_leave_event_delay_ms; - const retryInterval = matrixRtcSessionConfig?.network_error_retry_ms; - + // Set maximumNetworkErrorRetryCount such that we will consider the client + // disconnected as soon as either it fails to sync for longer than the grace + // period, or it is likely that a delayed leave event has been sent. // Math.min is used to account for the respective worst case: /sync not available or leave event emitted. - const maxWaitTime = Math.min(gracePeriod, leaveDelay); + const maxWaitTime = Math.min(gracePeriod, delayedLeaveTimings.delay_ms); const maximumNetworkErrorRetryCount = Math.ceil(maxWaitTime / retryInterval) + 1; @@ -936,18 +998,15 @@ export function enterRTCSession( notificationType, callIntent, manageMediaKeys: encryptMedia, - delayedLeaveEventRestartMs: - matrixRtcSessionConfig?.delayed_leave_event_restart_ms, - delayedLeaveEventDelayMs: - matrixRtcSessionConfig?.delayed_leave_event_delay_ms, + delayedLeaveEventRestartMs: delayedLeaveTimings.restart_ms, + delayedLeaveEventDelayMs: delayedLeaveTimings.delay_ms, delayedLeaveEventRestartLocalTimeoutMs: - matrixRtcSessionConfig?.delayed_leave_event_restart_local_timeout_ms, - networkErrorRetryMs: matrixRtcSessionConfig?.network_error_retry_ms, - makeKeyDelay: matrixRtcSessionConfig?.wait_for_key_rotation_ms, - membershipEventExpiryMs: - matrixRtcSessionConfig?.membership_event_expiry_ms, + delayedLeaveTimings.restart_timeout_ms, + networkErrorRetryMs: sessionConfig?.network_error_retry_ms, + makeKeyDelay: sessionConfig?.wait_for_key_rotation_ms, + membershipEventExpiryMs: sessionConfig?.membership_event_expiry_ms, keyRotationParticipantLimit: - matrixRtcSessionConfig?.key_rotation_participant_limit, + sessionConfig?.key_rotation_participant_limit, unstableSendStickyEvents: matrixRTCMode === MatrixRTCMode.Matrix_2_0, maximumNetworkErrorRetryCount: maximumNetworkErrorRetryCount, }, diff --git a/src/utils/observable.test.ts b/src/utils/observable.test.ts index 80cbb3c83..392f742ab 100644 --- a/src/utils/observable.test.ts +++ b/src/utils/observable.test.ts @@ -9,9 +9,30 @@ import { expect, test } from "vitest"; import { type Observable, of, Subject, switchMap } from "rxjs"; import { withTestScheduler } from "./test"; -import { filterBehavior, generateItems, pauseWhen } from "./observable"; +import { or$, filterBehavior, generateItems, pauseWhen } from "./observable"; import { type Behavior } from "../state/Behavior"; +const yesNo = { + y: true, + n: false, +}; + +test("or$", () => { + withTestScheduler(({ behavior, expectObservable }) => { + const input1Marbles = "ny--n--"; + const input2Marbles = "n-y--n-"; + const input3Marbles = "n--y--n"; + const outputMarbles = "nyyyyyn"; + expectObservable( + or$( + behavior(input1Marbles, yesNo), + behavior(input2Marbles, yesNo), + behavior(input3Marbles, yesNo), + ), + ).toBe(outputMarbles, yesNo); + }); +}); + test("pauseWhen", () => { withTestScheduler(({ behavior, expectObservable }) => { const inputMarbles = " abcdefgh-i-jk-"; diff --git a/src/utils/observable.ts b/src/utils/observable.ts index c32254db3..d4ea90b22 100644 --- a/src/utils/observable.ts +++ b/src/utils/observable.ts @@ -114,13 +114,11 @@ export function getValue(state$: Observable): T { } /** - * Creates an Observable that has a value of true whenever all its inputs are - * true. - * - * @public + * Creates an Observable that has a value of true whenever some of its inputs + * are true. */ -export function and$(...inputs: Observable[]): Observable { - return combineLatest(inputs, (...flags) => flags.every((flag) => flag)); +export function or$(...inputs: Observable[]): Observable { + return combineLatest(inputs, (...flags) => flags.some((flag) => flag)); } /** diff --git a/vite-embedded.config.ts b/vite-embedded.config.ts index 27a42fbbf..228d2cf8f 100644 --- a/vite-embedded.config.ts +++ b/vite-embedded.config.ts @@ -27,8 +27,6 @@ export default defineConfig((env) => data: { matrix_rtc_session: { wait_for_key_rotation_ms: 5000, - delayed_leave_event_restart_ms: 4000, - delayed_leave_event_delay_ms: 18000, }, }, }, From 839e36abd41c3d8ab2cf6d784c7d1f55307a5f2a Mon Sep 17 00:00:00 2001 From: Robin Date: Tue, 8 Sep 2026 21:21:36 +0200 Subject: [PATCH 2/6] Make delegation support check fail immediately in case of CORS errors --- src/state/CallViewModel/localMember/LocalMember.ts | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/src/state/CallViewModel/localMember/LocalMember.ts b/src/state/CallViewModel/localMember/LocalMember.ts index 838b14ee2..fc5e65cf2 100644 --- a/src/state/CallViewModel/localMember/LocalMember.ts +++ b/src/state/CallViewModel/localMember/LocalMember.ts @@ -75,7 +75,6 @@ import { import { type HomeserverConnected } from "./HomeserverConnected.ts"; import { type LocalTransport } from "./LocalTransport.ts"; import { areLivekitTransportsEqual } from "../remoteMembers/MatrixLivekitMembers.ts"; -import { doNetworkOperationWithRetry } from "../../../utils/matrix.ts"; import { or$ } from "../../../utils/observable.ts"; export enum TransportState { @@ -253,10 +252,12 @@ export const createLocalMembership$ = ({ ): Promise { logger.info(`Checking whether ${serviceName} supports delegation…`); try { - // Bluntly hit the endpoint without auth to check for a 404 - const res = await doNetworkOperationWithRetry(async () => - fetch(endpointUrl, { method: "POST" }), - ); + // Bluntly hit the endpoint without auth to check for a 404. Unfortunately + // we can't wrap this in a retry loop, as many servers don't just disable + // delegation support, but in fact are from a time before the endpoint + // existed at all, therefore we can hit CORS errors which would just gum + // up the retry loop. (May be revisited after Matrix 2.0.) + const res = await fetch(endpointUrl, { method: "POST" }); if (res.status === 404) { logger.warn(`${serviceName} does not support delegation`); return false; From 09782a4d8497e6c3a6047a9eb06d5c14ac5d28c3 Mon Sep 17 00:00:00 2001 From: Robin Date: Tue, 8 Sep 2026 15:41:23 +0200 Subject: [PATCH 3/6] Give JWT token fetch procedures more appropriate names --- src/livekit/openIDSFU.ts | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/src/livekit/openIDSFU.ts b/src/livekit/openIDSFU.ts index 67dd5e9e7..785d80c7c 100644 --- a/src/livekit/openIDSFU.ts +++ b/src/livekit/openIDSFU.ts @@ -126,9 +126,9 @@ export async function getSFUConfigWithOpenID( if (tryBothJwtEndpoints || forceMatrix2Jwt) { try { logger?.info( - `Trying to get JWT with delegation for focus ${serviceUrl}...`, + `Trying to get JWT via default endpoint for focus ${serviceUrl}...`, ); - const sfuConfig = await getLiveKitJWTWithDelayDelegation( + const sfuConfig = await getLiveKitJWT( membership, serviceUrl, roomId, @@ -154,7 +154,7 @@ export async function getSFUConfigWithOpenID( logger?.info( `Trying to get JWT with legacy endpoint for focus ${serviceUrl}...`, ); - sfuConfig = await getLiveKitJWT( + sfuConfig = await getLiveKitJWTLegacy( membership.deviceId, serviceUrl, roomId, @@ -188,7 +188,7 @@ function extractFullConfigFromToken(sfuConfig: { }; } -async function getLiveKitJWT( +async function getLiveKitJWTLegacy( deviceId: string, livekitServiceURL: string, matrixRoomId: string, @@ -263,7 +263,7 @@ class NotSupportedError extends Error { } } -export async function getLiveKitJWTWithDelayDelegation( +export async function getLiveKitJWT( membership: CallMembershipIdentityParts, livekitServiceURL: string, matrixRoomId: string, From d38daef56736779b06e1f5f51cf560b2e2306bee Mon Sep 17 00:00:00 2001 From: Robin Date: Tue, 8 Sep 2026 15:42:50 +0200 Subject: [PATCH 4/6] Reflect session config always being defined --- src/state/CallViewModel/localMember/LocalMember.ts | 9 ++++----- 1 file changed, 4 insertions(+), 5 deletions(-) diff --git a/src/state/CallViewModel/localMember/LocalMember.ts b/src/state/CallViewModel/localMember/LocalMember.ts index fc5e65cf2..36c69e6ed 100644 --- a/src/state/CallViewModel/localMember/LocalMember.ts +++ b/src/state/CallViewModel/localMember/LocalMember.ts @@ -1003,11 +1003,10 @@ export function enterRTCSession( delayedLeaveEventDelayMs: delayedLeaveTimings.delay_ms, delayedLeaveEventRestartLocalTimeoutMs: delayedLeaveTimings.restart_timeout_ms, - networkErrorRetryMs: sessionConfig?.network_error_retry_ms, - makeKeyDelay: sessionConfig?.wait_for_key_rotation_ms, - membershipEventExpiryMs: sessionConfig?.membership_event_expiry_ms, - keyRotationParticipantLimit: - sessionConfig?.key_rotation_participant_limit, + networkErrorRetryMs: sessionConfig.network_error_retry_ms, + makeKeyDelay: sessionConfig.wait_for_key_rotation_ms, + membershipEventExpiryMs: sessionConfig.membership_event_expiry_ms, + keyRotationParticipantLimit: sessionConfig.key_rotation_participant_limit, unstableSendStickyEvents: matrixRTCMode === MatrixRTCMode.Matrix_2_0, maximumNetworkErrorRetryCount: maximumNetworkErrorRetryCount, }, From 5cfe7ea4dfc290ac13d117435a1a656a37f9e487 Mon Sep 17 00:00:00 2001 From: Robin Date: Tue, 8 Sep 2026 20:50:26 +0200 Subject: [PATCH 5/6] Base JWT service endpoint choice directly on the MatrixRTC mode Since we removed the extra 'legacy' mode, meaning there's nothing simplified anymore by translating things over to a JWT service version. (Also, simplify some things by making the mode static for the duration of a call.) --- src/livekit/openIDSFU.ts | 14 ++-- src/room/InCallView.tsx | 4 +- src/state/CallViewModel/CallViewModel.ts | 68 +++++++------------ .../CallViewModel/CallViewModelTestUtils.ts | 2 +- .../localMember/LocalMember.test.ts | 1 + .../CallViewModel/localMember/LocalMember.ts | 21 +++--- .../localMember/LocalTransport.test.ts | 11 +-- .../localMember/LocalTransport.ts | 18 ++--- src/utils/observable.ts | 32 --------- src/utils/test-viewmodel.ts | 2 +- 10 files changed, 58 insertions(+), 115 deletions(-) diff --git a/src/livekit/openIDSFU.ts b/src/livekit/openIDSFU.ts index 785d80c7c..cf1f3c951 100644 --- a/src/livekit/openIDSFU.ts +++ b/src/livekit/openIDSFU.ts @@ -19,7 +19,7 @@ import { } from "../utils/errors"; import { doNetworkOperationWithRetry } from "../utils/matrix"; import { Config } from "../config/Config"; -import { JwtEndpointVersion } from "../state/CallViewModel/localMember/LocalTransport"; +import { MatrixRTCMode } from "../config/ConfigOptions"; /** * Configuration and access tokens provided by the SFU on successful authentication. @@ -80,11 +80,10 @@ export type OpenIDClientParts = Pick< * @param serviceUrl The URL of the livekit SFU service * @param roomId The room id used in the jwt request. This is NOT the livekit_alias. The jwt service will provide the alias. It maps matrix room ids <-> Livekit aliases. * @param opts Additional options to modify which endpoint with which data will be used to acquire the jwt token. - * @param opts.forceJwtEndpoint This will use the old jwt endpoint which will create the rtc backend identity based on string concatenation - * instead of a hash. + * @param opts.matrixRTCMode Determines which version of the JWT endpoint to use, which affects whether the + * RTC backend identity is based on string concatenation (legacy) or a hash (Matrix 2.0). * This function by default uses whatever is possible with the current jwt service installed next to the SFU. * For remote connections this does not matter, since we will not publish there we can rely on the newest option. - * For our own connection we can only use the hashed version if we also send the new matrix2.0 sticky events. * @param opts.delayEndpointBaseUrl The URL of the matrix homeserver. * @param opts.delayId The delay id used for the jwt service to manage. * @param logger optional logger. @@ -97,7 +96,7 @@ export async function getSFUConfigWithOpenID( serviceUrl: string, roomId: string, opts?: { - forceJwtEndpoint?: JwtEndpointVersion; + matrixRTCMode?: MatrixRTCMode; delayEndpointBaseUrl?: string; delayId?: string; }, @@ -116,10 +115,9 @@ export async function getSFUConfigWithOpenID( logger?.debug("Got openID token", openIdToken); let sfuConfig: { url: string; jwt: string } | undefined; - const tryBothJwtEndpoints = opts?.forceJwtEndpoint === undefined; // This is for SFUs where we do not publish. + const tryBothJwtEndpoints = opts?.matrixRTCMode === undefined; // This is for SFUs where we do not publish. - const forceMatrix2Jwt = - opts?.forceJwtEndpoint === JwtEndpointVersion.Matrix_2_0; + const forceMatrix2Jwt = opts?.matrixRTCMode === MatrixRTCMode.Matrix_2_0; // We want to start using the new endpoint (with optional delay delegation) // if we can use both or if we are forced to use the new one. diff --git a/src/room/InCallView.tsx b/src/room/InCallView.tsx index 3ea88f81f..85fe66a74 100644 --- a/src/room/InCallView.tsx +++ b/src/room/InCallView.tsx @@ -135,7 +135,9 @@ export const ActiveCall: FC = (props) => { encryptionSystem: props.e2eeSystem, autoLeaveWhenOthersLeft, waitForCallPickup: waitForCallPickup && sendNotificationType === "ring", - matrixRTCMode$: matrixRTCModeSetting.value$, + // We merely sample the current mode here, so the user would need to + // manually rejoin to switch to a different one. + matrixRTCMode: matrixRTCModeSetting.value$.value, }, reactionsReader.raisedHands$, reactionsReader.reactions$, diff --git a/src/state/CallViewModel/CallViewModel.ts b/src/state/CallViewModel/CallViewModel.ts index 62f05a476..0bd33f213 100644 --- a/src/state/CallViewModel/CallViewModel.ts +++ b/src/state/CallViewModel/CallViewModel.ts @@ -54,7 +54,6 @@ import { type IMembershipManager } from "matrix-js-sdk/lib/matrixrtc/IMembership import { createToggle$, filterBehavior, - generateItem, generateItems, pauseWhen, } from "../../utils/observable"; @@ -113,7 +112,6 @@ import { } from "./localMember/LocalMember.ts"; import { createLocalTransport$, - JwtEndpointVersion, type LocalTransport, } from "./localMember/LocalTransport.ts"; import { @@ -191,7 +189,7 @@ export interface CallViewModelOptions { /** Optional value overriding the connection factory, for testing purposes. */ connectionFactory?: ConnectionFactory; /** The version & compatibility mode of MatrixRTC that we should use. */ - matrixRTCMode$?: Behavior; + matrixRTCMode?: MatrixRTCMode; /** Optional behavior overriding for the screensharing, for testing */ toggleScreensharing?: () => void; } @@ -453,10 +451,8 @@ export function createCallViewModel$( const configMatrixRTCMode = Config.get().matrix_rtc_mode as | MatrixRTCMode | undefined; - const matrixRTCMode$ = - configMatrixRTCMode !== undefined - ? constant(configMatrixRTCMode) - : (options.matrixRTCMode$ ?? constant(MatrixRTCMode.Compatibility)); + const matrixRTCMode = + configMatrixRTCMode ?? options.matrixRTCMode ?? MatrixRTCMode.Compatibility; // Each hbar seperates a block of input variables required for the CallViewModel to function. // The outputs of this block is written under the hbar. @@ -490,38 +486,26 @@ export function createCallViewModel$( memberId: uuidv4(), }; - const localTransport$ = scope.behavior( - matrixRTCMode$.pipe( - generateItem( - "CallViewModel localTransport$", - // Re-create LocalTransport whenever the mode changes - (mode) => ({ keys: [mode], data: undefined }), - (scope, _data$, mode) => - options.localTransport ?? - createLocalTransport$({ - scope: scope, - memberships$: memberships$, - ownMembershipIdentity, - client, - delayId$: scope.behavior( - ( - fromEvent( - matrixRTCSession, - MembershipManagerEvent.DelayIdChanged, - // The type of reemitted event includes the original emitted as the second arg. - ) as Observable<[string | undefined, IMembershipManager]> - ).pipe(map(([delayId]) => delayId ?? null)), - matrixRTCSession.delayId ?? null, - ), - roomId: matrixRoom.roomId, - forceJwtEndpoint: - mode === MatrixRTCMode.Matrix_2_0 - ? JwtEndpointVersion.Matrix_2_0 - : JwtEndpointVersion.Legacy, - }), + const localTransport = + options.localTransport ?? + createLocalTransport$({ + scope: scope, + memberships$: memberships$, + ownMembershipIdentity, + client, + delayId$: scope.behavior( + ( + fromEvent( + matrixRTCSession, + MembershipManagerEvent.DelayIdChanged, + // The type of reemitted event includes the original emitted as the second arg. + ) as Observable<[string | undefined, IMembershipManager]> + ).pipe(map(([delayId]) => delayId ?? null)), + matrixRTCSession.delayId ?? null, ), - ), - ); + roomId: matrixRoom.roomId, + matrixRTCMode, + }); const connectionFactory = options.connectionFactory ?? @@ -539,8 +523,7 @@ export function createCallViewModel$( scope: scope, connectionFactory: connectionFactory, localTransport$: scope.behavior( - localTransport$.pipe( - switchMap((t) => t.active$), + localTransport.active$.pipe( catchError((e: unknown) => { logger.info( "could not pass local transport to createConnectionManager$. localTransport$ threw an error", @@ -583,9 +566,7 @@ export function createCallViewModel$( transport, { encryptMedia: livekitKeyProvider !== undefined, - // We merely sample the current mode here, so the user would need to - // manually rejoin to switch to a different one - matrixRTCMode: matrixRTCMode$.value, + matrixRTCMode, delayedLeaveTimings, }, ); @@ -606,6 +587,7 @@ export function createCallViewModel$( localTransport$, roomId: matrixRoom.roomId, baseUrl: client.baseUrl, + matrixRTCMode, logger: logger.getChild(`[${Date.now()}]`), }); diff --git a/src/state/CallViewModel/CallViewModelTestUtils.ts b/src/state/CallViewModel/CallViewModelTestUtils.ts index 2e1525c06..5a3fa3378 100644 --- a/src/state/CallViewModel/CallViewModelTestUtils.ts +++ b/src/state/CallViewModel/CallViewModelTestUtils.ts @@ -238,7 +238,7 @@ export function withCallViewModel(mode: MatrixRTCMode) { ); }, }, - matrixRTCMode$: constant(mode), + matrixRTCMode: mode, ...options, }, raisedHands$, diff --git a/src/state/CallViewModel/localMember/LocalMember.test.ts b/src/state/CallViewModel/localMember/LocalMember.test.ts index 66cd8cda0..6fe2155bd 100644 --- a/src/state/CallViewModel/localMember/LocalMember.test.ts +++ b/src/state/CallViewModel/localMember/LocalMember.test.ts @@ -230,6 +230,7 @@ describe("LocalMembership", () => { }, roomId: "!test-room-id:example.org", baseUrl: "https://matrix.example.org", + matrixRTCMode: MATRIX_RTC_MODE, }; beforeEach(() => { diff --git a/src/state/CallViewModel/localMember/LocalMember.ts b/src/state/CallViewModel/localMember/LocalMember.ts index 36c69e6ed..9b9218c54 100644 --- a/src/state/CallViewModel/localMember/LocalMember.ts +++ b/src/state/CallViewModel/localMember/LocalMember.ts @@ -143,7 +143,7 @@ interface Props { ) => void; homeserverConnected: HomeserverConnected; roomId: string; - localTransport$: Behavior; + localTransport: LocalTransport; matrixRTCSession: Pick< MatrixRTCSession, "updateCallIntent" | "leaveRoomSession" @@ -163,7 +163,7 @@ interface Props { * @param props.createPublisherFactory Factory to create a publisher once we have a connection. * @param props.joinMatrixRTC Callback to join the matrix RTC session once we have a transport. * @param props.homeserverConnected The homeserver connected state. - * @param props.localTransport$ The transport to advertise in our membership. + * @param props.localTransport The transport to advertise in our membership. * @param props.logger The logger to use. * @param props.muteStates The mute states for video and audio. * @param props.matrixRTCSession The matrix RTC session to join. @@ -282,8 +282,7 @@ export const createLocalMembership$ = ({ // The transport that we will advertise in our membership, paired with info as // to whether delayed event delegation is supported - const joinParams$ = localTransport$.pipe( - switchMap((lt) => lt.advertised$), + const joinParams$ = localTransport.advertised$.pipe( catchError(handleTransportError), distinctUntilChanged(areLivekitTransportsEqual), switchMap((transport) => { @@ -304,16 +303,12 @@ export const createLocalMembership$ = ({ // Unwrap the local transport and set the state of the LocalMembership to error in case the transport is an error. const activeTransport$ = scope.behavior( - localTransport$.pipe( - switchMap((lt) => { - return combineLatest([lt.active$, lt.advertised$]).pipe( - map(([active, advertised]) => { - // Our policy is to not publish to another transport if our prefered transport is miss-configured - if (advertised == null) return null; + combineLatest([localTransport.active$, localTransport.advertised$]).pipe( + map(([active, advertised]) => { + // Our policy is to not publish to another transport if our prefered transport is miss-configured + if (advertised == null) return null; - return active?.transport ?? null; - }), - ); + return active?.transport ?? null; }), catchError(handleTransportError), distinctUntilChanged(areLivekitTransportsEqual), diff --git a/src/state/CallViewModel/localMember/LocalTransport.test.ts b/src/state/CallViewModel/localMember/LocalTransport.test.ts index 09f6ecec0..7f9ecb98a 100644 --- a/src/state/CallViewModel/localMember/LocalTransport.test.ts +++ b/src/state/CallViewModel/localMember/LocalTransport.test.ts @@ -41,6 +41,7 @@ import { import * as openIDSFU from "../../../livekit/openIDSFU"; import { customLivekitUrl } from "../../../settings/settings"; import { testJWTToken } from "../../../utils/test-fixtures"; +import { MatrixRTCMode } from "../../../config/ConfigOptions"; describe("LocalTransport", () => { const openIdResponse: openIDSFU.SFUConfig = { @@ -67,7 +68,7 @@ describe("LocalTransport", () => { getDeviceId: vi.fn(), }, ownMembershipIdentity: ownMemberMock, - forceJwtEndpoint: JwtEndpointVersion.Legacy, + matrixRTCMode: MatrixRTCMode.Compatibility, delayId$: constant("delay_id_mock"), }); await flushPromises(); @@ -108,7 +109,7 @@ describe("LocalTransport", () => { getDeviceId: vi.fn(), }, ownMembershipIdentity: ownMemberMock, - forceJwtEndpoint: JwtEndpointVersion.Legacy, + matrixRTCMode: MatrixRTCMode.Compatibility, delayId$: constant("delay_id_mock"), }); active$.subscribe( @@ -150,7 +151,7 @@ describe("LocalTransport", () => { baseUrl: "https://example.org", }, ownMembershipIdentity: ownMemberMock, - forceJwtEndpoint: JwtEndpointVersion.Legacy, + matrixRTCMode: MatrixRTCMode.Compatibility, delayId$: constant("delay_id_mock"), }); @@ -194,7 +195,7 @@ describe("LocalTransport", () => { ownMembershipIdentity: ownMemberMock, scope: testScope(), roomId: "!example_room_id", - forceJwtEndpoint: JwtEndpointVersion.Legacy, + matrixRTCMode: MatrixRTCMode.Compatibility, delayId$: constant(null), memberships$: constant(new Epoch([])), client: { @@ -306,7 +307,7 @@ describe("LocalTransport", () => { scope: testScope(), ownMembershipIdentity: ownMemberMock, roomId: "!example_room_id", - forceJwtEndpoint: JwtEndpointVersion.Legacy, + matrixRTCMode: MatrixRTCMode.Compatibility, delayId$: constant(null), memberships$: constant(new Epoch([])), client: { diff --git a/src/state/CallViewModel/localMember/LocalTransport.ts b/src/state/CallViewModel/localMember/LocalTransport.ts index f98a266fd..66be511a4 100644 --- a/src/state/CallViewModel/localMember/LocalTransport.ts +++ b/src/state/CallViewModel/localMember/LocalTransport.ts @@ -37,6 +37,7 @@ import { import { areLivekitTransportsEqual } from "../remoteMembers/MatrixLivekitMembers.ts"; import { customLivekitUrl } from "../../../settings/settings.ts"; import { RtcTransportAutoDiscovery } from "./RtcTransportAutoDiscovery.ts"; +import { type MatrixRTCMode } from "../../../config/ConfigOptions.ts"; /* * It figures out “which LiveKit focus URL/alias the local user should use,” @@ -53,15 +54,10 @@ interface Props { OpenIDClientParts; // Used by the jwt service to create the livekit room and compute the livekit alias. roomId: string; - forceJwtEndpoint: JwtEndpointVersion; + matrixRTCMode: MatrixRTCMode; delayId$: Behavior; } -export enum JwtEndpointVersion { - Legacy = "legacy", - Matrix_2_0 = "matrix_2_0", -} - // TODO livekit_alias-cleanup // 1. We need to move away from transports map to connections!!! // @@ -122,7 +118,7 @@ export const createLocalTransport$ = ({ ownMembershipIdentity, client, roomId, - forceJwtEndpoint, + matrixRTCMode, delayId$, }: Props): LocalTransport => { const logger = rootLogger.getChild("[LocalTransport]"); @@ -167,7 +163,7 @@ export const createLocalTransport$ = ({ try { return await doOpenIdAndJWTFromUrl( transport, - forceJwtEndpoint, + matrixRTCMode, ownMembershipIdentity, roomId, client, @@ -219,7 +215,7 @@ export const createLocalTransport$ = ({ * use we don't want to risk any issues by re-using a token. * * @param transport The transport to authenticate with. - * @param forceJwtEndpoint Whether to force the JWT endpoint to be used. + * @param matrixRTCMode Whether to force the JWT endpoint to be used. * @param membership The identity of the local member. * @param roomId The room ID to use for the JWT. * @param client The client to use for the OpenID token. @@ -229,7 +225,7 @@ export const createLocalTransport$ = ({ */ async function doOpenIdAndJWTFromUrl( transport: LivekitTransportConfig, - forceJwtEndpoint: JwtEndpointVersion, + matrixRTCMode: MatrixRTCMode, membership: CallMembershipIdentityParts, roomId: string, client: Pick< @@ -246,7 +242,7 @@ async function doOpenIdAndJWTFromUrl( transport.livekit_service_url, roomId, { - forceJwtEndpoint: forceJwtEndpoint, + matrixRTCMode, delayEndpointBaseUrl: client.baseUrl, delayId, }, diff --git a/src/utils/observable.ts b/src/utils/observable.ts index d4ea90b22..8468b3fbb 100644 --- a/src/utils/observable.ts +++ b/src/utils/observable.ts @@ -227,38 +227,6 @@ export function filterBehavior( ); } -/** - * Maps a changing input value to an item whose lifetime is tied to a certain - * computed key. The item may capture some dynamic data from the input. - */ -export function generateItem< - Input, - Keys extends [unknown, ...unknown[]], - Data, - Item, ->( - name: string, - generator: (input: Input) => { keys: readonly [...Keys]; data: Data }, - factory: ( - scope: ObservableScope, - data$: Behavior, - ...keys: Keys - ) => Item, -): OperatorFunction { - return (input$) => - input$.pipe( - generateItemsInternal( - name, - function* (input) { - yield generator(input); - }, - factory, - (items) => items, - ), - map(([item]) => item), - ); -} - function generateItemsInternal< Input, Keys extends [unknown, ...unknown[]], diff --git a/src/utils/test-viewmodel.ts b/src/utils/test-viewmodel.ts index f53910024..974879110 100644 --- a/src/utils/test-viewmodel.ts +++ b/src/utils/test-viewmodel.ts @@ -174,7 +174,7 @@ export function getBasicCallViewModelEnvironment( setE2EEEnabled: async () => Promise.resolve(), }), connectionState$: constant(ConnectionState.Connected), - matrixRTCMode$: constant(MatrixRTCMode.Compatibility), + matrixRTCMode: MatrixRTCMode.Compatibility, ...callViewModelOptions, }, handRaisedSubject$, From cff8766aa618f8350eb98a71aa49df6c19719847 Mon Sep 17 00:00:00 2001 From: Robin Date: Tue, 8 Sep 2026 20:54:27 +0200 Subject: [PATCH 6/6] Delegate delayed leaves in LocalMember rather than LocalTransport IMO this is where the delegation calls should have lived all along, since the leave event is part of the membership lifecycle, and we otherwise end up with an awkward hack to ignore transport updates. Doing this now ensures that the client won't send any delegation requests if delegation is unsupported, and prepares the code for a future change in which we use the dedicated delegation endpoint from the CS API. --- src/state/CallViewModel/CallViewModel.ts | 24 +-- .../localMember/LocalMember.test.ts | 149 +++++++++++------- .../CallViewModel/localMember/LocalMember.ts | 41 ++++- .../localMember/LocalTransport.test.ts | 103 +----------- .../localMember/LocalTransport.ts | 94 ++++------- 5 files changed, 178 insertions(+), 233 deletions(-) diff --git a/src/state/CallViewModel/CallViewModel.ts b/src/state/CallViewModel/CallViewModel.ts index 0bd33f213..aa88e6115 100644 --- a/src/state/CallViewModel/CallViewModel.ts +++ b/src/state/CallViewModel/CallViewModel.ts @@ -493,16 +493,6 @@ export function createCallViewModel$( memberships$: memberships$, ownMembershipIdentity, client, - delayId$: scope.behavior( - ( - fromEvent( - matrixRTCSession, - MembershipManagerEvent.DelayIdChanged, - // The type of reemitted event includes the original emitted as the second arg. - ) as Observable<[string | undefined, IMembershipManager]> - ).pipe(map(([delayId]) => delayId ?? null)), - matrixRTCSession.delayId ?? null, - ), roomId: matrixRoom.roomId, matrixRTCMode, }); @@ -583,10 +573,22 @@ export function createCallViewModel$( ); }, connectionManager, + client, matrixRTCSession, - localTransport$, + localTransport, roomId: matrixRoom.roomId, baseUrl: client.baseUrl, + ownMembershipIdentity, + delayId$: scope.behavior( + ( + fromEvent( + matrixRTCSession, + MembershipManagerEvent.DelayIdChanged, + // The type of reemitted event includes the original emitted as the second arg. + ) as Observable<[string | undefined, IMembershipManager]> + ).pipe(map(([delayId]) => delayId ?? null)), + matrixRTCSession.delayId ?? null, + ), matrixRTCMode, logger: logger.getChild(`[${Date.now()}]`), }); diff --git a/src/state/CallViewModel/localMember/LocalMember.test.ts b/src/state/CallViewModel/localMember/LocalMember.test.ts index 6fe2155bd..e7c5b1e45 100644 --- a/src/state/CallViewModel/localMember/LocalMember.test.ts +++ b/src/state/CallViewModel/localMember/LocalMember.test.ts @@ -64,6 +64,7 @@ import { type LocalTransport, type LocalTransportWithSFUConfig, } from "./LocalTransport"; +import * as openIDSFU from "../../../livekit/openIDSFU"; initializeWidget(); @@ -113,6 +114,17 @@ const delegatedTimings: ResolvedDelayedLeaveTimings = { restart_timeout_ms: timings.restart_timeout_ms! * 10, }; +const mockedClient = { + getDomain: vi.fn().mockReturnValue("example.org"), + getDeviceId: vi.fn().mockReturnValue("AAAA"), + getOpenIdToken: vi.fn().mockResolvedValue({ + access_token: "ACCCESS_TOKEN", + token_type: "Bearer", + matrix_server_name: "localhost", + expires_in: 10000, + }), +}; + describe("enterRTCSession", () => { const transport: LivekitTransportConfig = { livekit_alias: "roomId", @@ -129,15 +141,7 @@ describe("enterRTCSession", () => { const mockedSession = vi.mocked({ room: { roomId: "roomId", - client: { - getDomain: vi.fn().mockReturnValue("example.org"), - getOpenIdToken: vi.fn().mockResolvedValue({ - access_token: "ACCCESS_TOKEN", - token_type: "Bearer", - matrix_server_name: "localhost", - expires_in: 10000, - }), - }, + client: mockedClient, }, memberships: [], joinRTCSession: vi.fn(), @@ -230,6 +234,9 @@ describe("LocalMembership", () => { }, roomId: "!test-room-id:example.org", baseUrl: "https://matrix.example.org", + ownMembershipIdentity: ownMemberMock, + client: mockedClient, + delayId$: constant(null), matrixRTCMode: MATRIX_RTC_MODE, }; @@ -276,7 +283,7 @@ describe("LocalMembership", () => { scope, ...defaultCreateLocalMemberValues, connectionManager: mockConnectionManager, - localTransport$: behavior("a", { a: aLocalTransport }), + localTransport: aLocalTransport, }); expectObservable(localMembership.localMemberState$).toBe("ne", { @@ -320,7 +327,7 @@ describe("LocalMembership", () => { scope, ...defaultCreateLocalMemberValues, connectionManager: mockConnectionManager, - localTransport$: constant(aLocalTransport), + localTransport: aLocalTransport, }); expectObservable(localMembership.localMemberState$).toBe("n-e", { @@ -337,8 +344,8 @@ describe("LocalMembership", () => { const scope = new ObservableScope(); const aLocalTransport: LocalTransport = { - advertised$: new BehaviorSubject(aTransport), - active$: new BehaviorSubject(aTransportWithSFUConfig), + advertised$: constant(aTransport), + active$: constant(aTransportWithSFUConfig), }; const mockConnectionManager = { @@ -356,7 +363,7 @@ describe("LocalMembership", () => { leaveRoomSession: vi.fn(), }, connectionManager: mockConnectionManager, - localTransport$: new BehaviorSubject(aLocalTransport), + localTransport: aLocalTransport, }); const expextedLog = "'not connected yet' while updating the call intent (this is expected on startup)"; @@ -417,6 +424,11 @@ describe("LocalMembership", () => { livekitRoom: mockLivekitRoom({}), } as unknown as Connection; + const authCallSpy = vi + .spyOn(openIDSFU, "getSFUConfigWithOpenID") + .mockImplementation(() => mockedClient.getOpenIdToken()); + afterEach(() => authCallSpy.mockClear()); + it.each([ ["no", null, timings], [ @@ -429,14 +441,8 @@ describe("LocalMembership", () => { "joins session with %s delegation support", async (_serviceName, delegationUrl, delayedLeaveTimings) => { const scope = testScope(); - - const activeTransport$ = constant(aTransportWithSFUConfig); - const aLocalTransport: LocalTransport = { - advertised$: constant(aTransport), - active$: activeTransport$, - }; - const connectionManagerData = new ConnectionManagerData(); const joinMatrixRTC = vi.fn(); + const delayId$ = new BehaviorSubject(null); if (delegationUrl !== null) fetchMock.post(delegationUrl, () => ({ status: 401, body: {} })); @@ -445,10 +451,16 @@ describe("LocalMembership", () => { scope, ...defaultCreateLocalMemberValues, connectionManager: { - connectionManagerData$: constant(new Epoch(connectionManagerData)), + connectionManagerData$: constant( + new Epoch(new ConnectionManagerData()), + ), }, - localTransport$: constant(aLocalTransport), joinMatrixRTC, + localTransport: { + advertised$: constant(aTransport), + active$: constant(aTransportWithSFUConfig), + }, + delayId$, }); localMembership.requestJoinAndPublish(); @@ -459,6 +471,34 @@ describe("LocalMembership", () => { aTransport, delayedLeaveTimings, ); + + expect(authCallSpy).not.toHaveBeenCalled(); + delayId$.next("leave1"); + await flushPromises(); + if (delegationUrl === null) { + expect(authCallSpy).not.toHaveBeenCalled(); + } else { + // Delegation is supported in this test case, so go on to check that + // LocalMember actually performs delegation + const expectDelegation = (delayId: string) => + expect(authCallSpy).toHaveBeenLastCalledWith( + mockedClient, + ownMemberMock, + "a", + "!test-room-id:example.org", + { + matrixRTCMode: MATRIX_RTC_MODE, + delayEndpointBaseUrl: "https://matrix.example.org", + delayId, + }, + expect.anything(), + ); + + expectDelegation("leave1"); + delayId$.next("leave2"); // Can change delegated leaves + await flushPromises(); + expectDelegation("leave2"); + } }, ); @@ -467,7 +507,7 @@ describe("LocalMembership", () => { const activeTransport$ = new BehaviorSubject(aTransportWithSFUConfig); const aLocalTransport: LocalTransport = { - advertised$: new BehaviorSubject(aTransport), + advertised$: constant(aTransport), active$: activeTransport$, }; @@ -505,7 +545,7 @@ describe("LocalMembership", () => { connectionManager: { connectionManagerData$: constant(new Epoch(connectionManagerData)), }, - localTransport$: new BehaviorSubject(aLocalTransport), + localTransport: aLocalTransport, }); await flushPromises(); activeTransport$.next({ @@ -536,7 +576,7 @@ describe("LocalMembership", () => { const publishers: Publisher[] = []; const tracks$ = new BehaviorSubject([]); - const publishing$ = new BehaviorSubject(false); + const publishing$ = constant(false); defaultCreateLocalMemberValues.createPublisherFactory.mockImplementation( () => { const p = { @@ -560,8 +600,8 @@ describe("LocalMembership", () => { >; const aLocalTransport: LocalTransport = { - advertised$: new BehaviorSubject(aTransport), - active$: new BehaviorSubject(aTransportWithSFUConfig), + advertised$: constant(aTransport), + active$: constant(aTransportWithSFUConfig), }; const connectionManagerData = new ConnectionManagerData(); @@ -573,7 +613,7 @@ describe("LocalMembership", () => { connectionManager: { connectionManagerData$: constant(new Epoch(connectionManagerData)), }, - localTransport$: new BehaviorSubject(aLocalTransport), + localTransport: aLocalTransport, }); await flushPromises(); expect(publisherFactory).toHaveBeenCalledOnce(); @@ -601,7 +641,7 @@ describe("LocalMembership", () => { new BehaviorSubject(null); const aLocalTransport: LocalTransport = { - advertised$: new BehaviorSubject(aTransport), + advertised$: constant(aTransport), active$: activeTransport$, }; @@ -644,7 +684,7 @@ describe("LocalMembership", () => { connectionManager: { connectionManagerData$, }, - localTransport$: new BehaviorSubject(aLocalTransport), + localTransport: aLocalTransport, }); await flushPromises(); @@ -779,10 +819,10 @@ describe("LocalMembership", () => { connectionManager: { connectionManagerData$: constant(new Epoch(connectionManagerData)), }, - localTransport$: new BehaviorSubject({ - advertised$: new BehaviorSubject(aTransport), - active$: new BehaviorSubject(aTransportWithSFUConfig), - }), + localTransport: { + advertised$: constant(aTransport), + active$: constant(aTransportWithSFUConfig), + }, }); await flushPromises(); @@ -819,10 +859,10 @@ describe("LocalMembership", () => { connectionManager: { connectionManagerData$: constant(new Epoch(connectionManagerData)), }, - localTransport$: new BehaviorSubject({ - advertised$: new BehaviorSubject(aTransport), - active$: new BehaviorSubject(aTransportWithSFUConfig), - }), + localTransport: { + advertised$: constant(aTransport), + active$: constant(aTransportWithSFUConfig), + }, }); await flushPromises(); @@ -861,18 +901,19 @@ describe("LocalMembership", () => { scope, ...defaultCreateLocalMemberValues, homeserverConnected: { - combined$: new BehaviorSubject< - [boolean, HomeserverDisconnectReason | null] - >([true, null]), + combined$: constant<[boolean, HomeserverDisconnectReason | null]>([ + true, + null, + ]), rtsSession$: constant(RTCMemberStatus.Connected), }, connectionManager: { connectionManagerData$: constant(new Epoch(connectionManagerData)), }, - localTransport$: new BehaviorSubject({ - advertised$: new BehaviorSubject(aTransport), - active$: new BehaviorSubject(aTransportWithSFUConfig), - }), + localTransport: { + advertised$: constant(aTransport), + active$: constant(aTransportWithSFUConfig), + }, }); await flushPromises(); @@ -913,10 +954,10 @@ describe("LocalMembership", () => { connectionManager: { connectionManagerData$: constant(new Epoch(connectionManagerData)), }, - localTransport$: new BehaviorSubject({ - advertised$: new BehaviorSubject(aTransport), - active$: new BehaviorSubject(aTransportWithSFUConfig), - }), + localTransport: { + advertised$: constant(aTransport), + active$: constant(aTransportWithSFUConfig), + }, }); await flushPromises(); @@ -982,10 +1023,10 @@ describe("LocalMembership", () => { connectionManager: { connectionManagerData$: constant(new Epoch(connectionManagerData)), }, - localTransport$: new BehaviorSubject({ - advertised$: new BehaviorSubject(aTransport), - active$: new BehaviorSubject(aTransportWithSFUConfig), - }), + localTransport: { + advertised$: constant(aTransport), + active$: constant(aTransportWithSFUConfig), + }, }); return { scope, localMembership }; }; diff --git a/src/state/CallViewModel/localMember/LocalMember.ts b/src/state/CallViewModel/localMember/LocalMember.ts index 9b9218c54..7805805b8 100644 --- a/src/state/CallViewModel/localMember/LocalMember.ts +++ b/src/state/CallViewModel/localMember/LocalMember.ts @@ -15,6 +15,7 @@ import { MediaDeviceFailure, } from "livekit-client"; import { observeParticipantEvents } from "@livekit/components-core"; +import { type MatrixClient } from "matrix-js-sdk"; import { Status as RTCSessionStatus, type LivekitTransport, @@ -76,6 +77,7 @@ import { type HomeserverConnected } from "./HomeserverConnected.ts"; import { type LocalTransport } from "./LocalTransport.ts"; import { areLivekitTransportsEqual } from "../remoteMembers/MatrixLivekitMembers.ts"; import { or$ } from "../../../utils/observable.ts"; +import { getSFUConfigWithOpenID } from "../../../livekit/openIDSFU.ts"; export enum TransportState { /** Not even a transport is available to the LocalMembership */ @@ -143,12 +145,16 @@ interface Props { ) => void; homeserverConnected: HomeserverConnected; roomId: string; + ownMembershipIdentity: CallMembershipIdentityParts; localTransport: LocalTransport; + client: Pick; matrixRTCSession: Pick< MatrixRTCSession, "updateCallIntent" | "leaveRoomSession" >; baseUrl: string; + delayId$: Behavior; + matrixRTCMode: MatrixRTCMode; logger: Logger; } @@ -168,6 +174,7 @@ interface Props { * @param props.muteStates The mute states for video and audio. * @param props.matrixRTCSession The matrix RTC session to join. * @param props.baseUrl Base URL of the homeserver. + * @param props.delayId$ ID of the delayed leave event to delegate to the SFU. * @param props.roomId The room ID used as the call identifier in analytics events. * @returns * - publisher: The handle to create tracks and publish them to the room. @@ -179,15 +186,19 @@ interface Props { export const createLocalMembership$ = ({ scope, connectionManager, - localTransport$, + localTransport, homeserverConnected, createPublisherFactory, joinMatrixRTC, logger: parentLogger, muteStates, + client, matrixRTCSession, baseUrl, roomId, + ownMembershipIdentity, + delayId$, + matrixRTCMode, }: Props): { /** * This request to start audio and video tracks. @@ -711,6 +722,34 @@ export const createLocalMembership$ = ({ ), ); + // Delegate delayed leaves to the SFU + scope.reconcile( + scope.behavior(combineLatest([joinParams$, delayId$])), + async ([joinParams, delayId]) => { + if (joinParams?.delegationSupported && delayId !== null) { + try { + // This will technically cause the service to issue a new JWT token, + // but it's safe to discard. We're only interested in triggering + // delegation. + await getSFUConfigWithOpenID( + client, + ownMembershipIdentity, + joinParams.transport.livekit_service_url, + roomId, + { matrixRTCMode, delayEndpointBaseUrl: baseUrl, delayId }, + logger, + ); + } catch (e) { + // TODO: Surface this to the user as a service interruption? + logger.error( + `Failed to delegate leave to ${joinParams.transport.livekit_service_url}`, + e, + ); + } + } + }, + ); + // Pause upstream of all local media tracks when we're disconnected from // MatrixRTC, because it can be an unpleasant surprise for the app to say // 'reconnecting' and yet still be transmitting your media to others. diff --git a/src/state/CallViewModel/localMember/LocalTransport.test.ts b/src/state/CallViewModel/localMember/LocalTransport.test.ts index 7f9ecb98a..a77c7a16d 100644 --- a/src/state/CallViewModel/localMember/LocalTransport.test.ts +++ b/src/state/CallViewModel/localMember/LocalTransport.test.ts @@ -14,11 +14,8 @@ import { type MockedObject, vi, } from "vitest"; -import { - type CallMembership, - type LivekitTransportConfig, -} from "matrix-js-sdk/lib/matrixrtc"; -import { BehaviorSubject, filter, lastValueFrom } from "rxjs"; +import { type CallMembership } from "matrix-js-sdk/lib/matrixrtc"; +import { lastValueFrom } from "rxjs"; import fetchMock from "fetch-mock"; import { @@ -27,11 +24,7 @@ import { ownMemberMock, testScope, } from "../../../utils/test"; -import { - createLocalTransport$, - JwtEndpointVersion, - type LocalTransportWithSFUConfig, -} from "./LocalTransport"; +import { createLocalTransport$ } from "./LocalTransport"; import { constant } from "../../Behavior"; import { Epoch, ObservableScope } from "../../ObservableScope"; import { @@ -62,14 +55,12 @@ describe("LocalTransport", () => { // eslint-disable-next-line @typescript-eslint/naming-convention _unstable_getRTCTransports: async () => Promise.resolve([]), getDomain: () => "example.org", - baseUrl: "example.org", // These won't be called in this error path but satisfy the type getOpenIdToken: vi.fn(), getDeviceId: vi.fn(), }, ownMembershipIdentity: ownMemberMock, matrixRTCMode: MatrixRTCMode.Compatibility, - delayId$: constant("delay_id_mock"), }); await flushPromises(); @@ -101,7 +92,6 @@ describe("LocalTransport", () => { roomId: "!example_room_id", memberships$: constant(new Epoch([])), client: { - baseUrl: "https://example.org", getDomain: () => "example.org", // eslint-disable-next-line @typescript-eslint/naming-convention _unstable_getRTCTransports: async () => Promise.resolve([]), @@ -110,7 +100,6 @@ describe("LocalTransport", () => { }, ownMembershipIdentity: ownMemberMock, matrixRTCMode: MatrixRTCMode.Compatibility, - delayId$: constant("delay_id_mock"), }); active$.subscribe( (o) => observations.push(o), @@ -148,11 +137,9 @@ describe("LocalTransport", () => { getDomain: () => "example.org", getOpenIdToken: vi.fn(), getDeviceId: vi.fn(), - baseUrl: "https://example.org", }, ownMembershipIdentity: ownMemberMock, matrixRTCMode: MatrixRTCMode.Compatibility, - delayId$: constant("delay_id_mock"), }); openIdResolver.resolve?.({ @@ -196,10 +183,8 @@ describe("LocalTransport", () => { scope: testScope(), roomId: "!example_room_id", matrixRTCMode: MatrixRTCMode.Compatibility, - delayId$: constant(null), memberships$: constant(new Epoch([])), client: { - baseUrl: "https://example.org", getDomain: vi.fn().mockReturnValue("example.org"), // eslint-disable-next-line @typescript-eslint/naming-convention _unstable_getRTCTransports: vi.fn().mockResolvedValue([]), @@ -308,11 +293,9 @@ describe("LocalTransport", () => { ownMembershipIdentity: ownMemberMock, roomId: "!example_room_id", matrixRTCMode: MatrixRTCMode.Compatibility, - delayId$: constant(null), memberships$: constant(new Epoch([])), client: { getDomain: () => "example.org", - baseUrl: "https://example.org", // eslint-disable-next-line @typescript-eslint/naming-convention _unstable_getRTCTransports: async () => Promise.resolve([]), // These won't be called in this error path but satisfy the type @@ -330,84 +313,4 @@ describe("LocalTransport", () => { ); }); }); - - it("should not update advertised/active transport on delayID changes, but delay Id delegation should be called", async () => { - // For simplicity, we'll just use the config livekit - customLivekitUrl.setValue("https://lk.example.org"); - - const authCallSpy = vi - .spyOn(openIDSFU, "getSFUConfigWithOpenID") - .mockResolvedValue(openIdResponse); - - const delayId$ = new BehaviorSubject(null); - - const { advertised$, active$ } = createLocalTransport$({ - scope: testScope(), - ownMembershipIdentity: ownMemberMock, - roomId: "!example_room_id", - // We want multi-sdu - forceJwtEndpoint: JwtEndpointVersion.Legacy, - delayId$: delayId$, - memberships$: constant(new Epoch([])), - client: { - getDomain: () => "example.org", - baseUrl: "https://example.org", - // eslint-disable-next-line @typescript-eslint/naming-convention - _unstable_getRTCTransports: async () => Promise.resolve([]), - // These won't be called in this error path but satisfy the type - getOpenIdToken: vi.fn(), - getDeviceId: vi.fn(), - }, - }); - - const advertisedValues: LivekitTransportConfig[] = []; - const activeValues: LocalTransportWithSFUConfig[] = []; - advertised$ - .pipe(filter((v) => v !== null)) - .subscribe((t) => advertisedValues.push(t)); - active$ - .pipe(filter((v) => v !== null)) - .subscribe((t) => activeValues.push(t)); - - await flushPromises(); - - // we have now an active and an advertised - expect(advertisedValues.length).toEqual(1); - expect(activeValues.length).toEqual(1); - expect(advertisedValues[0]!.livekit_service_url).toEqual( - "https://lk.example.org", - ); - expect(activeValues[0]!.transport.livekit_service_url).toEqual( - "https://lk.example.org", - ); - - expect(authCallSpy).toHaveBeenCalledTimes(2); - // Now emits 3 new delays id - delayId$.next("delay_id_1"); - await flushPromises(); - delayId$.next("delay_id_2"); - await flushPromises(); - delayId$.next("delay_id_3"); - await flushPromises(); - - // No new emissions should've happened, it is the same transport. - expect(advertisedValues.length).toEqual(1); - expect(activeValues.length).toEqual(1); - - // Still we should have updated the delayID to auth - expect(authCallSpy).toHaveBeenCalledTimes( - 4 * 2 /* 2 calls for each delayId ?? why */, - ); - - expect(authCallSpy).toHaveBeenLastCalledWith( - expect.anything(), - expect.anything(), - expect.anything(), - expect.anything(), - expect.objectContaining({ - delayId: "delay_id_3", - }), - expect.anything(), - ); - }); }); diff --git a/src/state/CallViewModel/localMember/LocalTransport.ts b/src/state/CallViewModel/localMember/LocalTransport.ts index 66be511a4..c5aa39590 100644 --- a/src/state/CallViewModel/localMember/LocalTransport.ts +++ b/src/state/CallViewModel/localMember/LocalTransport.ts @@ -10,14 +10,7 @@ import { type LivekitTransportConfig, } from "matrix-js-sdk/lib/matrixrtc"; import { type MatrixClient } from "matrix-js-sdk"; -import { - combineLatest, - distinctUntilChanged, - from, - map, - of, - switchMap, -} from "rxjs"; +import { distinctUntilChanged, from, map, of, switchMap } from "rxjs"; import { logger as rootLogger, type Logger } from "matrix-js-sdk/lib/logger"; import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager"; @@ -47,15 +40,11 @@ interface Props { scope: ObservableScope; ownMembershipIdentity: CallMembershipIdentityParts; memberships$: Behavior>; - client: Pick< - MatrixClient, - "getDomain" | "baseUrl" | "_unstable_getRTCTransports" - > & + client: Pick & OpenIDClientParts; // Used by the jwt service to create the livekit room and compute the livekit alias. roomId: string; matrixRTCMode: MatrixRTCMode; - delayId$: Behavior; } // TODO livekit_alias-cleanup @@ -119,7 +108,6 @@ export const createLocalTransport$ = ({ client, roomId, matrixRTCMode, - delayId$, }: Props): LocalTransport => { const logger = rootLogger.getChild("[LocalTransport]"); @@ -134,32 +122,29 @@ export const createLocalTransport$ = ({ transportDiscovery.discoverPreferredTransport(), ); - const preferredConfig$ = customLivekitUrl.value$ - .pipe( - switchMap((customUrl) => { - if (customUrl) { - return of({ - type: "livekit", - livekit_service_url: customUrl, - } as LivekitTransportConfig); - } else { - return discoveredTransport$; - } - }), - ) - .pipe( - map((config) => { - if (!config) { - // Bubbled up from the preferredConfig$ observable. - throw new MatrixRTCTransportMissingError(client.getDomain() ?? ""); - } - return config; - }), - distinctUntilChanged(areLivekitTransportsEqual), - ); + const preferredConfig$ = customLivekitUrl.value$.pipe( + switchMap((customUrl) => { + if (customUrl) { + return of({ + type: "livekit", + livekit_service_url: customUrl, + } as LivekitTransportConfig); + } else { + return discoveredTransport$; + } + }), + map((config) => { + if (!config) { + // Bubbled up from the preferredConfig$ observable. + throw new MatrixRTCTransportMissingError(client.getDomain() ?? ""); + } + return config; + }), + distinctUntilChanged(areLivekitTransportsEqual), + ); - const preferredTransport$ = combineLatest([preferredConfig$, delayId$]).pipe( - switchMap(async ([transport, delayId]) => { + const preferredTransport$ = preferredConfig$.pipe( + switchMap(async (transport) => { try { return await doOpenIdAndJWTFromUrl( transport, @@ -167,7 +152,6 @@ export const createLocalTransport$ = ({ ownMembershipIdentity, roomId, client, - delayId ?? undefined, logger, ); } catch (e) { @@ -189,21 +173,7 @@ export const createLocalTransport$ = ({ ), null, ), - active$: scope.behavior( - preferredTransport$.pipe( - // XXX: WORK AROUND due to a reconnection glitch. - // To remove when we have a proper way to refresh the delegation event ID without refreshing - // the whole credentials. - // We deliberately hide any changes to the SFU config because we - // do not want the app to reconnect whenever the JWT - // token changes due to us delegating a new delayed event. The - // initial SFU config for the transport is all the app needs. - distinctUntilChanged((prev, next) => - areLivekitTransportsEqual(prev.transport, next.transport), - ), - ), - null, - ), + active$: scope.behavior(preferredTransport$, null), }; }; @@ -219,7 +189,6 @@ export const createLocalTransport$ = ({ * @param membership The identity of the local member. * @param roomId The room ID to use for the JWT. * @param client The client to use for the OpenID token. - * @param delayId The delayId to use for the JWT. * * @throws FailToGetOpenIdToken, NoMatrix2AuthorizationService */ @@ -228,12 +197,7 @@ async function doOpenIdAndJWTFromUrl( matrixRTCMode: MatrixRTCMode, membership: CallMembershipIdentityParts, roomId: string, - client: Pick< - MatrixClient, - "getDomain" | "baseUrl" | "_unstable_getRTCTransports" - > & - OpenIDClientParts, - delayId?: string, + client: Pick & OpenIDClientParts, logger?: Logger, ): Promise { const sfuConfig = await getSFUConfigWithOpenID( @@ -241,11 +205,7 @@ async function doOpenIdAndJWTFromUrl( membership, transport.livekit_service_url, roomId, - { - matrixRTCMode, - delayEndpointBaseUrl: client.baseUrl, - delayId, - }, + { matrixRTCMode }, logger, ); return {