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..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,19 +115,18 @@ 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. 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 +152,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 +186,7 @@ function extractFullConfigFromToken(sfuConfig: { }; } -async function getLiveKitJWT( +async function getLiveKitJWTLegacy( deviceId: string, livekitServiceURL: string, matrixRoomId: string, @@ -204,11 +202,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, }; } @@ -264,7 +261,7 @@ class NotSupportedError extends Error { } } -export async function getLiveKitJWTWithDelayDelegation( +export async function getLiveKitJWT( membership: CallMembershipIdentityParts, livekitServiceURL: string, matrixRoomId: string, @@ -288,11 +285,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/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 c854363be..aa88e6115 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"; @@ -64,7 +63,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"; @@ -110,7 +112,6 @@ import { } from "./localMember/LocalMember.ts"; import { createLocalTransport$, - JwtEndpointVersion, type LocalTransport, } from "./localMember/LocalTransport.ts"; import { @@ -188,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; } @@ -450,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. @@ -487,38 +486,16 @@ 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, + roomId: matrixRoom.roomId, + matrixRTCMode, + }); const connectionFactory = options.connectionFactory ?? @@ -536,8 +513,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", @@ -562,16 +538,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 +546,19 @@ export function createCallViewModel$( matrixRTCSession, ), muteStates, - joinMatrixRTC: (transport: LivekitTransportConfig) => { + joinMatrixRTC: ( + transport: LivekitTransportConfig, + delayedLeaveTimings: ResolvedDelayedLeaveTimings, + ) => { return enterRTCSession( matrixRTCSession, ownMembershipIdentity, transport, - connectOptions$.value, + { + encryptMedia: livekitKeyProvider !== undefined, + matrixRTCMode, + delayedLeaveTimings, + }, ); }, createPublisherFactory: (connection: Connection) => { @@ -600,9 +573,23 @@ 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/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 aec641776..e7c5b1e45 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, @@ -58,6 +64,7 @@ import { type LocalTransport, type LocalTransportWithSFUConfig, } from "./LocalTransport"; +import * as openIDSFU from "../../../livekit/openIDSFU"; initializeWidget(); @@ -95,112 +102,112 @@ 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, - }, - ); +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, + }), +}; - 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 }), - ); - }); +describe("enterRTCSession", () => { + const transport: LivekitTransportConfig = { + livekit_alias: "roomId", + livekit_service_url: "http://my-livekit-service-url.com", + type: "livekit", + }; - 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 options = { + encryptMedia: true, + matrixRTCMode: MATRIX_RTC_MODE, + delayedLeaveTimings: timings, + }; - 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 mockedSession = vi.mocked({ + room: { + roomId: "roomId", + client: mockedClient, + }, + 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, - }, - ); + beforeEach(() => + mockConfig({ + livekit: { livekit_service_url: "http://my-default-service-url.com" }, + }), + ); - expect(mockedSession.joinRTCSession).toHaveBeenLastCalledWith( - expect.any(Object), - [], - expect.any(Object), - expect.objectContaining({ - keyRotationParticipantLimit: 50, - }), - ); - }); + 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 +233,30 @@ describe("LocalMembership", () => { rtsSession$: constant(RTCMemberStatus.Connected), }, roomId: "!test-room-id:example.org", + baseUrl: "https://matrix.example.org", + ownMembershipIdentity: ownMemberMock, + client: mockedClient, + delayId$: constant(null), + matrixRTCMode: MATRIX_RTC_MODE, }; + 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( @@ -254,9 +283,8 @@ describe("LocalMembership", () => { scope, ...defaultCreateLocalMemberValues, connectionManager: mockConnectionManager, - localTransport$: behavior("a", { a: aLocalTransport }), + localTransport: aLocalTransport, }); - localMembership.requestJoinAndPublish(); expectObservable(localMembership.localMemberState$).toBe("ne", { n: TransportState.Waiting, @@ -299,9 +327,8 @@ describe("LocalMembership", () => { scope, ...defaultCreateLocalMemberValues, connectionManager: mockConnectionManager, - localTransport$: behavior("a", { a: aLocalTransport }), + localTransport: aLocalTransport, }); - localMembership.requestJoinAndPublish(); expectObservable(localMembership.localMemberState$).toBe("n-e", { n: TransportState.Waiting, @@ -317,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 = { @@ -336,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)"; @@ -397,12 +424,90 @@ 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], + [ + "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 joinMatrixRTC = vi.fn(); + const delayId$ = new BehaviorSubject(null); + + if (delegationUrl !== null) + fetchMock.post(delegationUrl, () => ({ status: 401, body: {} })); + + const localMembership = createLocalMembership$({ + scope, + ...defaultCreateLocalMemberValues, + connectionManager: { + connectionManagerData$: constant( + new Epoch(new ConnectionManagerData()), + ), + }, + joinMatrixRTC, + localTransport: { + advertised$: constant(aTransport), + active$: constant(aTransportWithSFUConfig), + }, + delayId$, + }); + + localMembership.requestJoinAndPublish(); + void (await fetchMock.flush()); + await flushPromises(); + // Joins with timings appropriate for the level of delegation support + expect(joinMatrixRTC).toHaveBeenCalledWith( + 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"); + } + }, + ); + it("recreates publisher if new connection is used, always unpublish and end tracks", async () => { const scope = new ObservableScope(); const activeTransport$ = new BehaviorSubject(aTransportWithSFUConfig); const aLocalTransport: LocalTransport = { - advertised$: new BehaviorSubject(aTransport), + advertised$: constant(aTransport), active$: activeTransport$, }; @@ -440,7 +545,7 @@ describe("LocalMembership", () => { connectionManager: { connectionManagerData$: constant(new Epoch(connectionManagerData)), }, - localTransport$: new BehaviorSubject(aLocalTransport), + localTransport: aLocalTransport, }); await flushPromises(); activeTransport$.next({ @@ -471,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 = { @@ -495,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(); @@ -508,7 +613,7 @@ describe("LocalMembership", () => { connectionManager: { connectionManagerData$: constant(new Epoch(connectionManagerData)), }, - localTransport$: new BehaviorSubject(aLocalTransport), + localTransport: aLocalTransport, }); await flushPromises(); expect(publisherFactory).toHaveBeenCalledOnce(); @@ -536,7 +641,7 @@ describe("LocalMembership", () => { new BehaviorSubject(null); const aLocalTransport: LocalTransport = { - advertised$: new BehaviorSubject(aTransport), + advertised$: constant(aTransport), active$: activeTransport$, }; @@ -579,7 +684,7 @@ describe("LocalMembership", () => { connectionManager: { connectionManagerData$, }, - localTransport$: new BehaviorSubject(aLocalTransport), + localTransport: aLocalTransport, }); await flushPromises(); @@ -714,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(); @@ -754,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(); @@ -796,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(); @@ -848,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(); @@ -917,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 dfe8936a7..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, @@ -62,7 +63,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 +76,8 @@ import { 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 */ @@ -133,14 +139,22 @@ 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; + ownMembershipIdentity: CallMembershipIdentityParts; + localTransport: LocalTransport; + client: Pick; matrixRTCSession: Pick< MatrixRTCSession, "updateCallIntent" | "leaveRoomSession" >; + baseUrl: string; + delayId$: Behavior; + matrixRTCMode: MatrixRTCMode; logger: Logger; } @@ -155,10 +169,12 @@ 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. + * @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. @@ -170,14 +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. @@ -236,25 +257,69 @@ export const createLocalMembership$ = ({ return of(null); }; - // This is the transport that we will advertise in our membership. - const advertisedTransport$ = localTransport$.pipe( - switchMap((lt) => lt.advertised$), + 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. 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; + } 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.advertised$.pipe( 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. 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), @@ -617,16 +682,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) @@ -653,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. @@ -864,6 +961,7 @@ export function observeSharingScreen$(p: Participant): Observable { interface EnterRTCSessionOptions { encryptMedia: boolean; matrixRTCMode: MatrixRTCMode; + delayedLeaveTimings: ResolvedDelayedLeaveTimings; } /** @@ -876,6 +974,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 +983,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 +992,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 +1014,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 +1033,14 @@ 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, - keyRotationParticipantLimit: - matrixRtcSessionConfig?.key_rotation_participant_limit, + 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, unstableSendStickyEvents: matrixRTCMode === MatrixRTCMode.Matrix_2_0, maximumNetworkErrorRetryCount: maximumNetworkErrorRetryCount, }, diff --git a/src/state/CallViewModel/localMember/LocalTransport.test.ts b/src/state/CallViewModel/localMember/LocalTransport.test.ts index 09f6ecec0..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 { @@ -41,6 +34,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 = { @@ -61,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, - forceJwtEndpoint: JwtEndpointVersion.Legacy, - delayId$: constant("delay_id_mock"), + matrixRTCMode: MatrixRTCMode.Compatibility, }); await flushPromises(); @@ -100,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([]), @@ -108,8 +99,7 @@ describe("LocalTransport", () => { getDeviceId: vi.fn(), }, ownMembershipIdentity: ownMemberMock, - forceJwtEndpoint: JwtEndpointVersion.Legacy, - delayId$: constant("delay_id_mock"), + matrixRTCMode: MatrixRTCMode.Compatibility, }); active$.subscribe( (o) => observations.push(o), @@ -147,11 +137,9 @@ describe("LocalTransport", () => { getDomain: () => "example.org", getOpenIdToken: vi.fn(), getDeviceId: vi.fn(), - baseUrl: "https://example.org", }, ownMembershipIdentity: ownMemberMock, - forceJwtEndpoint: JwtEndpointVersion.Legacy, - delayId$: constant("delay_id_mock"), + matrixRTCMode: MatrixRTCMode.Compatibility, }); openIdResolver.resolve?.({ @@ -194,11 +182,9 @@ describe("LocalTransport", () => { ownMembershipIdentity: ownMemberMock, scope: testScope(), roomId: "!example_room_id", - forceJwtEndpoint: JwtEndpointVersion.Legacy, - delayId$: constant(null), + matrixRTCMode: MatrixRTCMode.Compatibility, 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([]), @@ -306,12 +292,10 @@ describe("LocalTransport", () => { scope: testScope(), ownMembershipIdentity: ownMemberMock, roomId: "!example_room_id", - forceJwtEndpoint: JwtEndpointVersion.Legacy, - delayId$: constant(null), + matrixRTCMode: MatrixRTCMode.Compatibility, 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 @@ -329,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 f98a266fd..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"; @@ -37,6 +30,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,” @@ -46,20 +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; - forceJwtEndpoint: JwtEndpointVersion; - delayId$: Behavior; -} - -export enum JwtEndpointVersion { - Legacy = "legacy", - Matrix_2_0 = "matrix_2_0", + matrixRTCMode: MatrixRTCMode; } // TODO livekit_alias-cleanup @@ -122,8 +107,7 @@ export const createLocalTransport$ = ({ ownMembershipIdentity, client, roomId, - forceJwtEndpoint, - delayId$, + matrixRTCMode, }: Props): LocalTransport => { const logger = rootLogger.getChild("[LocalTransport]"); @@ -138,40 +122,36 @@ 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, - forceJwtEndpoint, + matrixRTCMode, ownMembershipIdentity, roomId, client, - delayId ?? undefined, logger, ); } catch (e) { @@ -193,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,25 +185,19 @@ 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. - * @param delayId The delayId to use for the JWT. * * @throws FailToGetOpenIdToken, NoMatrix2AuthorizationService */ async function doOpenIdAndJWTFromUrl( transport: LivekitTransportConfig, - forceJwtEndpoint: JwtEndpointVersion, + 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( @@ -245,11 +205,7 @@ async function doOpenIdAndJWTFromUrl( membership, transport.livekit_service_url, roomId, - { - forceJwtEndpoint: forceJwtEndpoint, - delayEndpointBaseUrl: client.baseUrl, - delayId, - }, + { matrixRTCMode }, logger, ); return { 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..8468b3fbb 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)); } /** @@ -229,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$, 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, }, }, },