From 8638d4f159644e7a6b90b2358aaff831f1c14be1 Mon Sep 17 00:00:00 2001 From: "Timo K." Date: Wed, 14 Oct 2026 16:09:11 +0200 Subject: [PATCH] Fall back to publishing on a remote SFU if possible Port of edf1feae from johannes/rtc-remote-publish onto the SDK split, where the MatrixRTC logic lives in sdk/ rather than src/state/CallViewModel/. Where the original lands: - getLocalTransport's remote fallback: the LiveKit backend's new findRemotePublishTransport (sdk/src/media-backend/livekit/ RemotePublishTransport.ts), which tries the candidates from the js-sdk's getRemoteLivekitPublishCandidates oldest first and keeps the first whose homeserver issues a token with the publish grant. discoverLocalTransport asks the backend for it after the host's fallback, with a snapshot of the session's memberships from RTCParticipation, and throws MatrixRTCTransportMissingError where that finds nothing too. Transport discovery stays backend neutral: the MSC4195 token check is the backend's. - LocalMember's "no delegation on a remote SFU" and the custom transport's server_name: LivekitBackend.prepareLocalTransport locates the transport on the homeserver it names (serverNameOfTransport) and skips the delegation probe where that is not our own. - SessionBehaviors' membership.getTransportServerName: the same helper in LivekitBackend.mediaFor. - decodeSFUJWTPayload, SFUJWTPayload.video.canPublish and the exported isCsApiUnavailable: sdk/src/media-backend/livekit/auth/. - DeveloperSettingsTab, the widget capabilities, the locale string and the js-sdk pin (9405287c, 44.0.0): applied as is. The error boundary snapshots follow the string. Tests: the LocalTransport cases go to RemotePublishTransport.test.ts and LocalTransport.test.ts, the LocalMember remote case and the SessionBehaviors server-name case to LivekitBackend.test.ts, plus the memberships snapshot in RTCParticipation.test.ts. Co-Authored-By: Claude Fable 5.1 --- locales/en/app.json | 2 +- package.json | 2 +- pnpm-lock.yaml | 12 +- sdk/SdkArchitecture.md | 13 +- sdk/src/RTCParticipation.test.ts | 27 +++- sdk/src/RTCParticipation.ts | 18 ++- sdk/src/matrixrtc/LocalTransport.test.ts | 57 ++++++- sdk/src/matrixrtc/LocalTransport.ts | 19 ++- sdk/src/media-backend/api.ts | 16 +- .../livekit/LivekitBackend.test.ts | 68 ++++++++ .../media-backend/livekit/LivekitBackend.ts | 62 ++++++-- .../livekit/RemotePublishTransport.test.ts | 149 ++++++++++++++++++ .../livekit/RemotePublishTransport.ts | 90 +++++++++++ sdk/src/media-backend/livekit/auth/index.ts | 2 +- sdk/src/media-backend/livekit/auth/types.ts | 13 +- sdk/src/utils/serverName.ts | 16 ++ .../GroupCallErrorBoundary.test.tsx.snap | 6 +- src/settings/DeveloperSettingsTab.tsx | 2 +- src/widget.test.ts | 2 + src/widget.ts | 2 + 20 files changed, 527 insertions(+), 51 deletions(-) create mode 100644 sdk/src/media-backend/livekit/RemotePublishTransport.test.ts create mode 100644 sdk/src/media-backend/livekit/RemotePublishTransport.ts diff --git a/locales/en/app.json b/locales/en/app.json index 5be1bfff6..d57a4e8e3 100644 --- a/locales/en/app.json +++ b/locales/en/app.json @@ -106,7 +106,7 @@ "insufficient_capacity_description": "The server has reached its maximum capacity and you cannot join the call at this time. Try again later, or contact your server admin if the problem persists.", "livekit_connection_error": "Failed to connect to Livekit server", "livekit_connection_error_description": "An error occurred while connecting to the Livekit server (<1>Reason: <2>{{ reason }}).", - "matrix_rtc_transport_missing": "The server is not configured to work with {{brand}}. Please contact your server admin (Domain: {{domain}}, Error Code: {{ errorCode }}).", + "matrix_rtc_transport_missing": "The server is not configured to work with {{brand}} and no other call participant provides a usable transport. Please contact your server admin (Domain: {{domain}}, Error Code: {{ errorCode }}).", "membership_manager": "Membership Manager Error", "membership_manager_description": "The Membership Manager had to shut down. This is caused by many consecutive failed network requests.", "no_matrix_2_authorization_service": "The authorization service for your media server (SFU) is out of date.", diff --git a/package.json b/package.json index 089ed4b5a..a670d84d4 100644 --- a/package.json +++ b/package.json @@ -107,7 +107,7 @@ "livekit-client": "^2.18.1", "lodash-es": "^4.17.21", "loglevel": "^1.9.1", - "matrix-js-sdk": "github:matrix-org/matrix-js-sdk#develop", + "matrix-js-sdk": "github:matrix-org/matrix-js-sdk#9405287c22797e9d69acd366c47bf412fa4384c8", "matrix-widget-api": "^1.18.0", "node-stdlib-browser": "^1.3.1", "normalize.css": "^8.0.1", diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index cc0d93beb..ecafc61f5 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -191,8 +191,8 @@ importers: specifier: ^1.9.1 version: 1.9.2 matrix-js-sdk: - specifier: github:matrix-org/matrix-js-sdk#develop - version: https://codeload.github.com/matrix-org/matrix-js-sdk/tar.gz/4df2f2bfb5826c25f05d9535a8556acecaf2e570 + specifier: github:matrix-org/matrix-js-sdk#9405287c22797e9d69acd366c47bf412fa4384c8 + version: https://codeload.github.com/matrix-org/matrix-js-sdk/tar.gz/9405287c22797e9d69acd366c47bf412fa4384c8 matrix-widget-api: specifier: ^1.18.0 version: 1.20.0 @@ -4673,9 +4673,9 @@ packages: matrix-events-sdk@0.0.1: resolution: {integrity: sha512-1QEOsXO+bhyCroIe2/A5OwaxHvBm7EsSQ46DEDn8RBIfQwN5HWBpFvyWWR4QY0KHPPnnJdI99wgRiAl7Ad5qaA==} - matrix-js-sdk@https://codeload.github.com/matrix-org/matrix-js-sdk/tar.gz/4df2f2bfb5826c25f05d9535a8556acecaf2e570: - resolution: {gitHosted: true, integrity: sha512-b1o6ywaPY9jakj+3GIgx3samM9AbqohzPQ5nZKbyOGYB16IHQtNMyjEo/PY0fW/s0uwBVApEHNb/MMtN40c/rQ==, tarball: https://codeload.github.com/matrix-org/matrix-js-sdk/tar.gz/4df2f2bfb5826c25f05d9535a8556acecaf2e570} - version: 43.0.0 + matrix-js-sdk@https://codeload.github.com/matrix-org/matrix-js-sdk/tar.gz/9405287c22797e9d69acd366c47bf412fa4384c8: + resolution: {gitHosted: true, integrity: sha512-pYk38ggTEdM0WWXLDiFoKmyRn3wo3o8m1ge0iCtxjBmgAW5qoQ2Ifd8oHVy+gu6PQe/CfLuKPXEXH8hff9C8GA==, tarball: https://codeload.github.com/matrix-org/matrix-js-sdk/tar.gz/9405287c22797e9d69acd366c47bf412fa4384c8} + version: 44.0.0 engines: {node: '>=22.0.0'} matrix-widget-api@1.20.0: @@ -10227,7 +10227,7 @@ snapshots: matrix-events-sdk@0.0.1: {} - matrix-js-sdk@https://codeload.github.com/matrix-org/matrix-js-sdk/tar.gz/4df2f2bfb5826c25f05d9535a8556acecaf2e570: + matrix-js-sdk@https://codeload.github.com/matrix-org/matrix-js-sdk/tar.gz/9405287c22797e9d69acd366c47bf412fa4384c8: dependencies: '@babel/runtime': 8.0.0 '@matrix-org/matrix-sdk-crypto-wasm': 18.9.0 diff --git a/sdk/SdkArchitecture.md b/sdk/SdkArchitecture.md index be44ef4d6..4a184f6eb 100644 --- a/sdk/SdkArchitecture.md +++ b/sdk/SdkArchitecture.md @@ -101,6 +101,8 @@ LiveKit participants. │ create ───▶│ │ │ │ │ GET rtc/transports (session) ────────▶│ │ │ │◀── first transport of the backend's type │ + │ │ (none: findRemotePublishTransport, │ │ + │ │ a member's SFU we may publish on) ─▶│ │ │ │ prepareLocalTransport (backend): │ │ │ │ publisher token: CS API get_token, │ │ │ │ or OpenID token then /sfu/get ─────▶│ │ @@ -537,6 +539,10 @@ type MemberMediaFields = Pick; export interface MediaBackend { readonly transportType: string; // "livekit" + /** Another member's transport we may publish on, where the client found none. */ + findRemotePublishTransport( + memberships: CallMembership[], + ): Promise; /** Authenticate with the transport the client found; the probe for delegation runs here. */ prepareLocalTransport( transport: Transport, @@ -556,7 +562,9 @@ export interface MediaBackend { ``` What stays on the client's side: discovery of the local transport from the -homeserver's list (a homeserver endpoint, so not a backend's), the membership +homeserver's list (a homeserver endpoint, so not a backend's), asking the +backend for another member's transport only where that list and the host's +fallback are empty, the membership state machine in `LocalMembership.ts` and its combination with `local.connectionState$` into `state$`, the choice of delayed leave timings, mapping a data packet's sender to a member, and the transport registry, which keys on the raw @@ -566,7 +574,8 @@ transport serialised with sorted keys and derives `connectedTransports$` and eac The backend's context carries no Matrix client and no membership list. A backend that needs the client takes a `Pick` of exactly what it calls in its own options (`createLivekitBackend(scope, context, { client, mediaQuality })`), -and learns of remote members only through `mediaFor$`, registering each +and learns of remote members only through `mediaFor$` and the one membership +snapshot `findRemotePublishTransport` is handed, registering each membership's transport for as long as the member's scope lives. Media keys reach it as a stream, `mediaKeys$`, replayed on subscribe; the js-sdk session never does. A host supplies its own backend through `MatrixRTCClientOptions.backend`; diff --git a/sdk/src/RTCParticipation.test.ts b/sdk/src/RTCParticipation.test.ts index 6229068c2..48e63fbcc 100644 --- a/sdk/src/RTCParticipation.test.ts +++ b/sdk/src/RTCParticipation.test.ts @@ -84,6 +84,7 @@ interface FakeBackend { connectionState$: BehaviorSubject; data$: Subject; setPublishing: Mock<(publish: boolean) => void>; + findRemotePublishTransport: Mock<() => Promise>; delegateDelayedLeave: Mock<(delayId: string) => Promise>; mediaFor: Mock<(scope: ObservableScope, member: RTCMember) => MemberMedia>; } @@ -91,6 +92,7 @@ interface FakeBackend { function setup( options: Partial = {}, slotOptions: { matrixRTCMode?: MatrixRTCMode } = {}, + homeserverTransports: Transport[] = [transport], ): { scope: ObservableScope; participation: ReturnType["participation"]; @@ -112,7 +114,7 @@ function setup( getDeviceId: () => deviceId, getDomain: () => "example.org", getSyncState: () => SyncState.Syncing, - _unstable_getRTCTransports: vi.fn().mockResolvedValue([transport]), + _unstable_getRTCTransports: vi.fn().mockResolvedValue(homeserverTransports), } as unknown as MatrixClient; const members$ = new BehaviorSubject([]); @@ -124,6 +126,7 @@ function setup( ), data$: new Subject(), setPublishing: vi.fn(), + findRemotePublishTransport: vi.fn(async () => Promise.resolve(null)), delegateDelayedLeave: vi.fn(async () => Promise.resolve()), mediaFor: vi.fn(() => ({ tracks$: constant(null), @@ -136,6 +139,7 @@ function setup( ): MediaBackend => { const backend: MediaBackend = { transportType: "livekit", + findRemotePublishTransport: fake.findRemotePublishTransport, prepareLocalTransport: async () => fake.prepared.promise, delegateDelayedLeave: fake.delegateDelayedLeave, local: { @@ -220,6 +224,27 @@ describe("createRTCParticipation", () => { }); }); + it("publishes on another member's SFU where the homeserver advertises none", async () => { + const bobsTransport: Transport = { + type: "livekit", + url: "wss://sfu.bob.com", + server_name: "bob.com", + }; + const { session, join, fake } = setup({}, {}, []); + fake.findRemotePublishTransport.mockResolvedValue(bobsTransport); + fake.prepared.resolve({ canDelegateDelayedLeave: false }); + await flushPromises(); + + expect(fake.findRemotePublishTransport).toHaveBeenCalledWith( + session.memberships, + ); + expect(join).toHaveBeenCalledWith( + expect.anything(), + [bobsTransport], + expect.anything(), + ); + }); + it("publishes only while joined", async () => { const { participation, session, fake, scope } = setup(); const ended = vi.fn(); diff --git a/sdk/src/RTCParticipation.ts b/sdk/src/RTCParticipation.ts index adbafda38..31a39240a 100644 --- a/sdk/src/RTCParticipation.ts +++ b/sdk/src/RTCParticipation.ts @@ -131,12 +131,18 @@ export function createRTCParticipation( }); const preparedTransport$ = from( - discoverLocalTransport(client, backend.transportType, options, logger).then( - async (transport): Promise => ({ - transport, - ...(await backend.prepareLocalTransport(transport)), - }), - ), + discoverLocalTransport( + client, + backend, + // A snapshot: if nobody with a usable SFU is here yet, the user has to + // come back later rather than wait in a half-working lobby + jsSdkSession.memberships, + options, + logger, + ).then(async (transport): Promise => ({ + transport, + ...(await backend.prepareLocalTransport(transport)), + })), ); const localMembership = createLocalMembership$({ diff --git a/sdk/src/matrixrtc/LocalTransport.test.ts b/sdk/src/matrixrtc/LocalTransport.test.ts index 127f135a5..a2c834372 100644 --- a/sdk/src/matrixrtc/LocalTransport.test.ts +++ b/sdk/src/matrixrtc/LocalTransport.test.ts @@ -10,6 +10,7 @@ import { type Transport } from "matrix-js-sdk/lib/matrixrtc"; import { describe, expect, it, vi } from "vitest"; import { MatrixRTCTransportMissingError } from "../errors"; +import { mockRtcMembership } from "../utils/test"; import { discoverLocalTransport } from "./LocalTransport"; const homeserverTransport: Transport = { @@ -20,6 +21,12 @@ const hostTransport: Transport = { type: "livekit", livekit_service_url: "https://host.example.org", }; +const bobsTransport: Transport = { + type: "livekit", + url: "wss://sfu.bob.com", + server_name: "bob.com", +}; +const bob = mockRtcMembership("@bob:bob.com", "BOB"); function client( advertised: Transport[], @@ -30,6 +37,15 @@ function client( }; } +function backend( + remote: Transport | null = null, +): Parameters[1] { + return { + transportType: "livekit", + findRemotePublishTransport: vi.fn().mockResolvedValue(remote), + }; +} + describe("discoverLocalTransport", () => { it("takes the transport the host names without asking the homeserver", async () => { const c = client([homeserverTransport]); @@ -37,7 +53,8 @@ describe("discoverLocalTransport", () => { await expect( discoverLocalTransport( c, - "livekit", + backend(), + [], { transport: hostTransport }, logger, ), @@ -49,7 +66,8 @@ describe("discoverLocalTransport", () => { await expect( discoverLocalTransport( client([homeserverTransport]), - "livekit", + backend(), + [], { fallbackTransport: hostTransport }, logger, ), @@ -60,15 +78,46 @@ describe("discoverLocalTransport", () => { await expect( discoverLocalTransport( client([{ type: "not_livekit" }]), - "livekit", + backend(), + [], { fallbackTransport: hostTransport }, logger, ), ).resolves.toEqual(hostTransport); }); + it("prefers a local transport over other members' SFUs", async () => { + const b = backend(bobsTransport); + + await expect( + discoverLocalTransport( + client([homeserverTransport]), + b, + [bob], + {}, + logger, + ), + ).resolves.toEqual(homeserverTransport); + expect(b.findRemotePublishTransport).not.toHaveBeenCalled(); + }); + + it("publishes on another member's SFU where there is no local transport", async () => { + const b = backend(bobsTransport); + + await expect( + discoverLocalTransport(client([]), b, [bob], {}, logger), + ).resolves.toEqual(bobsTransport); + expect(b.findRemotePublishTransport).toHaveBeenCalledWith([bob]); + }); + it("fails naming the homeserver where there is no transport at all", async () => { - const promise = discoverLocalTransport(client([]), "livekit", {}, logger); + const promise = discoverLocalTransport( + client([]), + backend(), + [bob], + {}, + logger, + ); await expect(promise).rejects.toBeInstanceOf( MatrixRTCTransportMissingError, diff --git a/sdk/src/matrixrtc/LocalTransport.ts b/sdk/src/matrixrtc/LocalTransport.ts index 6815e85e1..09ffd1f9b 100644 --- a/sdk/src/matrixrtc/LocalTransport.ts +++ b/sdk/src/matrixrtc/LocalTransport.ts @@ -7,19 +7,27 @@ Please see LICENSE in the repository root for full details. import { type MatrixClient } from "matrix-js-sdk"; import { type Logger } from "matrix-js-sdk/lib/logger"; -import { type Transport } from "matrix-js-sdk/lib/matrixrtc"; +import { + type CallMembership, + type Transport, +} from "matrix-js-sdk/lib/matrixrtc"; import { type RTCParticipationOptions } from "../api"; import { MatrixRTCTransportMissingError } from "../errors"; +import { type MediaBackend } from "../media-backend/api"; import { TransportAutoDiscovery } from "./TransportAutoDiscovery"; /** * The transport the local member advertises: the one the host names, else the - * homeserver's first of the type the backend serves, else the host's fallback. + * homeserver's first of the type the backend serves, else the host's fallback, + * else another member's that the backend can publish on. + * + * @throws {MatrixRTCTransportMissingError} where none of these exists. */ export async function discoverLocalTransport( client: Pick, - transportType: string, + backend: Pick, + memberships: CallMembership[], { transport: override, fallbackTransport, @@ -30,10 +38,11 @@ export async function discoverLocalTransport( override ?? (await new TransportAutoDiscovery({ client, - transportType, + transportType: backend.transportType, fallbackTransport, logger, - }).discoverPreferredTransport()); + }).discoverPreferredTransport()) ?? + (await backend.findRemotePublishTransport(memberships)); if (transport === null) throw new MatrixRTCTransportMissingError(client.getDomain() ?? ""); return transport; diff --git a/sdk/src/media-backend/api.ts b/sdk/src/media-backend/api.ts index 4d5f5562f..b80277d00 100644 --- a/sdk/src/media-backend/api.ts +++ b/sdk/src/media-backend/api.ts @@ -13,7 +13,10 @@ Please see LICENSE in the repository root for full details. */ import { type Logger } from "matrix-js-sdk/lib/logger"; -import { type Transport } from "matrix-js-sdk/lib/matrixrtc"; +import { + type CallMembership, + type Transport, +} from "matrix-js-sdk/lib/matrixrtc"; import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager"; import { type Observable } from "rxjs"; @@ -116,6 +119,17 @@ export interface MediaBackend { /** The membership transport type this backend serves: "livekit". */ readonly transportType: string; + /** + * A transport to publish on where neither the homeserver nor the host + * provides one of the backend's type: for LiveKit, the SFU of another + * member whose homeserver lets us publish on it (MSC4195). Resolves null + * where no member's transport will do. Called at most once per session, + * with the memberships at that moment, before `prepareLocalTransport`. + */ + findRemotePublishTransport( + memberships: CallMembership[], + ): Promise; + /** * Makes `transport` usable as the one the local member publishes on. The * session found it; the backend does whatever has to succeed before it is diff --git a/sdk/src/media-backend/livekit/LivekitBackend.test.ts b/sdk/src/media-backend/livekit/LivekitBackend.test.ts index ab0b44e53..ddf60fd52 100644 --- a/sdk/src/media-backend/livekit/LivekitBackend.test.ts +++ b/sdk/src/media-backend/livekit/LivekitBackend.test.ts @@ -25,6 +25,7 @@ import { defaultSessionTimings } from "../../config"; import { flushPromises, mockEmitter, + mockRtcMembership, ownMemberMock, testScope, } from "../../utils/test"; @@ -258,6 +259,52 @@ describe("createLivekitBackend", () => { }, ); + it("prepares a remote SFU's transport on its homeserver, where delegation is not possible", async () => { + // Our own homeserver would delegate, so only the SFU being remote can + // explain delegation being skipped + fetchMock.post(homeserverEndpoint, 401); + const remoteTransport: UnstableLivekitTransport = { + type: "livekit", + url: "wss://sfu.sample.com", + server_name: "sample.com", + }; + + const capabilities = await backend().prepareLocalTransport(remoteTransport); + + expect(capabilities).toEqual({ canDelegateDelayedLeave: false }); + expect(connections[0].connection.role).toBe("publisher"); + expect(connections[0].connection.serverName).toBe("sample.com"); + }); + + it("finds another member's SFU to publish on through the homeserver", async () => { + client._unstable_getLivekitToken.mockResolvedValue({ + jwt: [ + {}, + { sub: "identity", video: { room: "room", canPublish: true } }, + {}, + ] + .map((d) => global.btoa(JSON.stringify(d))) + .join("."), + }); + const carl = mockRtcMembership("@carl:sample.com", "CARL", { + fociPreferred: [{ type: "livekit", url: "wss://sfu.sample.com" }], + }); + + await expect(backend().findRemotePublishTransport([carl])).resolves.toEqual( + { + type: "livekit", + url: "wss://sfu.sample.com", + server_name: "sample.com", + }, + ); + expect(client._unstable_getLivekitToken).toHaveBeenCalledWith( + expect.objectContaining({ + url: "wss://sfu.sample.com", + server_name: "sample.com", + }), + ); + }); + it("waits for the publisher connection to hold its token before the join", async () => { fetchMock.post(homeserverEndpoint, 404); fetchMock.post(`${exampleServiceUrl}/delegate_delayed_leave`, 404); @@ -488,6 +535,27 @@ describe("createLivekitBackend", () => { expect(remote?.connection.transport).toEqual(carlsTransport); expect(remote?.connection.serverName).toBe("sample.com"); }); + + it("connects to a remote publisher's transport on the SFU's homeserver", async () => { + const b = await prepared(); + // Dave has no SFU of his own and publishes on sample.com's + const remoteTransport: UnstableLivekitTransport = { + type: "livekit", + url: "wss://sfu.sample.com", + server_name: "sample.com", + }; + b.mediaFor(testScope(), { + ...bob(remoteTransport, ["dave-hashed"]), + userId: "@dave:example.org", + }); + await flushPromises(); + + const remote = connections.find( + (c) => c.connection.role === "subscriber", + ); + expect(remote?.connection.transport).toEqual(remoteTransport); + expect(remote?.connection.serverName).toBe("sample.com"); + }); }); }); diff --git a/sdk/src/media-backend/livekit/LivekitBackend.ts b/sdk/src/media-backend/livekit/LivekitBackend.ts index db7be819c..c139f6f52 100644 --- a/sdk/src/media-backend/livekit/LivekitBackend.ts +++ b/sdk/src/media-backend/livekit/LivekitBackend.ts @@ -8,6 +8,7 @@ Please see LICENSE in the repository root for full details. import { type MatrixClient } from "matrix-js-sdk"; import { type Logger } from "matrix-js-sdk/lib/logger"; import { + type CallMembership, isUnstableLivekitTransport, type Transport, } from "matrix-js-sdk/lib/matrixrtc"; @@ -34,7 +35,7 @@ import { trackEpoch, } from "../../reactive/ObservableScope"; import { mapScoped } from "../../utils/mapScoped"; -import { serverNameOf } from "../../utils/serverName"; +import { serverNameOf, serverNameOfTransport } from "../../utils/serverName"; import { type BackendConnection, type MediaBackend, @@ -62,6 +63,7 @@ import { createDataChannel$ } from "./DataChannel"; import { createKeyProvider } from "./KeyProvider"; import { createLivekitMemberMedia } from "./LivekitMemberMedia"; import { createLivekitLocalMedia } from "./LocalMedia"; +import { findRemotePublishTransport } from "./RemotePublishTransport"; export interface LivekitBackendOptions { /** @@ -146,26 +148,50 @@ export function createLivekitBackend( logger, ); + const ownServerName = serverNameOf(ownMembershipIdentity.userId); + + const remotePublishTransport = async ( + memberships: CallMembership[], + ): Promise => + findRemotePublishTransport({ + client, + memberships, + ownMembershipIdentity, + roomId, + slotId, + logger, + }); + const prepareLocalTransport = async ( transport: Transport, ): Promise => { if (!isUnstableLivekitTransport(transport)) throw new MatrixRTCError(`Not a LiveKit transport: ${transport.type}`); - const locator: TransportLocator = { + const serverName = serverNameOfTransport( transport, - serverName: serverNameOf(ownMembershipIdentity.userId), - }; - // Delegation goes through whichever service authenticates us as a - // publisher, so only that service's support counts - const flow = chooseAuthFlow(transport, "publisher"); - const canDelegate = - flow.kind === "openID" - ? supportsDelegation( - `${flow.serviceUrl}/delegate_delayed_leave`, - `transport ${flow.serviceUrl}`, - logger, - ) - : homeserverSupportsDelegation; + ownMembershipIdentity.userId, + ); + const locator: TransportLocator = { transport, serverName }; + let canDelegate: Promise; + if (serverName !== ownServerName) { + // Our homeserver only takes over leaves for its own SFUs + logger.info( + `Publishing on a remote SFU of ${serverName}, delegation is not possible`, + ); + canDelegate = Promise.resolve(false); + } else { + // Delegation goes through whichever service authenticates us as a + // publisher, so only that service's support counts + const flow = chooseAuthFlow(transport, "publisher"); + canDelegate = + flow.kind === "openID" + ? supportsDelegation( + `${flow.serviceUrl}/delegate_delayed_leave`, + `transport ${flow.serviceUrl}`, + logger, + ) + : homeserverSupportsDelegation; + } localTransport$.next(locator); // The membership is only sent once the publisher connection holds its // token, so that the LiveKit room exists before anyone subscribes to it: @@ -204,7 +230,10 @@ export function createLivekitBackend( map((transport): TransportLocator | undefined => { const raw = transport?.raw; return isUnstableLivekitTransport(raw) - ? { transport: raw, serverName: serverNameOf(member.userId) } + ? { + transport: raw, + serverName: serverNameOfTransport(raw, member.userId), + } : undefined; }), distinctUntilChanged((a, b) => (a && keyFor(a)) === (b && keyFor(b))), @@ -257,6 +286,7 @@ export function createLivekitBackend( return { transportType: "livekit", + findRemotePublishTransport: remotePublishTransport, prepareLocalTransport, delegateDelayedLeave: delegateLeave, local: localMedia_, diff --git a/sdk/src/media-backend/livekit/RemotePublishTransport.test.ts b/sdk/src/media-backend/livekit/RemotePublishTransport.test.ts new file mode 100644 index 000000000..455366af6 --- /dev/null +++ b/sdk/src/media-backend/livekit/RemotePublishTransport.test.ts @@ -0,0 +1,149 @@ +/* +Copyright 2026 Element Creations Ltd. + +SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial +Please see LICENSE in the repository root for full details. +*/ + +import { MatrixError } from "matrix-js-sdk"; +import { logger } from "matrix-js-sdk/lib/logger"; +import { type CallMembership } from "matrix-js-sdk/lib/matrixrtc"; +import { beforeEach, describe, expect, it, type Mocked, vi } from "vitest"; + +import { mockRtcMembership, ownMemberMock } from "../../utils/test"; +import { type ClientGetTokenParts } from "./auth/csApi"; +import { findRemotePublishTransport } from "./RemotePublishTransport"; + +/** A JWT whose payload states whether the issuing server lets us publish. */ +function jwtWithPublishGrant(canPublish: boolean): string { + return [ + {}, // header + { sub: "identity", video: { room: "room", canPublish } }, // payload + {}, // signature + ] + .map((d) => global.btoa(JSON.stringify(d))) + .join("."); +} + +const bobSfu = { type: "livekit", url: "wss://sfu.bob.com" }; +const carolSfu = { type: "livekit", url: "wss://sfu.carol.com" }; + +// Bob is the oldest member, Carol joined after him +const bob = mockRtcMembership("@bob:bob.com", "BOB", { + fociPreferred: [bobSfu], + membership: { created_ts: 100 }, +}); +const carol = mockRtcMembership("@carol:carol.com", "CAROL", { + fociPreferred: [carolSfu], + membership: { created_ts: 200 }, +}); +const legacyDave = mockRtcMembership("@dave:dave.com", "DAVE", { + fociPreferred: [ + { type: "livekit", livekit_service_url: "https://jwt.dave.com" }, + ], + membership: { created_ts: 50 }, +}); + +const csApiMissing = new MatrixError( + { errcode: "M_UNRECOGNIZED", error: "" }, + 404, +); + +describe("findRemotePublishTransport", () => { + let client: Mocked; + const find = async ( + memberships: CallMembership[], + ): ReturnType => + findRemotePublishTransport({ + client, + memberships, + ownMembershipIdentity: ownMemberMock, + roomId: "!room:example.org", + slotId: "m.call#ROOM", + logger, + }); + + beforeEach(() => { + client = { + _unstable_getLivekitToken: vi + .fn() + .mockResolvedValue({ jwt: jwtWithPublishGrant(true) }), + }; + }); + + it("publishes on the oldest member's SFU with its server_name", async () => { + await expect(find([carol, bob])).resolves.toEqual({ + type: "livekit", + url: "wss://sfu.bob.com", + server_name: "bob.com", + }); + expect(client._unstable_getLivekitToken).toHaveBeenCalledTimes(1); + expect(client._unstable_getLivekitToken).toHaveBeenCalledWith({ + url: "wss://sfu.bob.com", + server_name: "bob.com", + room_id: "!room:example.org", + slot_id: "m.call#ROOM", + member_id: ownMemberMock.memberId, + }); + }); + + it("ignores members publishing through a legacy JWT service", async () => { + await expect(find([legacyDave, bob])).resolves.toMatchObject({ + server_name: "bob.com", + }); + expect(client._unstable_getLivekitToken).toHaveBeenCalledTimes(1); + }); + + it("moves on when a homeserver does not let us publish", async () => { + client._unstable_getLivekitToken.mockImplementation( + async ({ server_name }) => + Promise.resolve({ + jwt: jwtWithPublishGrant(server_name === "carol.com"), + }), + ); + + await expect(find([bob, carol])).resolves.toMatchObject({ + server_name: "carol.com", + }); + expect(client._unstable_getLivekitToken).toHaveBeenCalledTimes(2); + }); + + it("moves on when a homeserver refuses to issue a token", async () => { + client._unstable_getLivekitToken.mockImplementation( + async ({ server_name }) => + server_name === "bob.com" + ? Promise.reject( + new MatrixError( + { errcode: "M_FORBIDDEN", error: "Not joined" }, + 403, + ), + ) + : Promise.resolve({ jwt: jwtWithPublishGrant(true) }), + ); + + await expect(find([bob, carol])).resolves.toMatchObject({ + server_name: "carol.com", + }); + }); + + it("finds nothing once all candidates are exhausted", async () => { + client._unstable_getLivekitToken.mockResolvedValue({ + jwt: jwtWithPublishGrant(false), + }); + + await expect(find([bob, carol])).resolves.toBeNull(); + expect(client._unstable_getLivekitToken).toHaveBeenCalledTimes(2); + }); + + it("gives up immediately if our homeserver lacks the CS API", async () => { + client._unstable_getLivekitToken.mockRejectedValue(csApiMissing); + + await expect(find([bob, carol])).resolves.toBeNull(); + expect(client._unstable_getLivekitToken).toHaveBeenCalledTimes(1); + }); + + it("finds nothing where there are no other members", async () => { + await expect(find([])).resolves.toBeNull(); + expect(client._unstable_getLivekitToken).not.toHaveBeenCalled(); + }); +}); diff --git a/sdk/src/media-backend/livekit/RemotePublishTransport.ts b/sdk/src/media-backend/livekit/RemotePublishTransport.ts new file mode 100644 index 000000000..988a6ed3d --- /dev/null +++ b/sdk/src/media-backend/livekit/RemotePublishTransport.ts @@ -0,0 +1,90 @@ +/* +Copyright 2026 Element Creations Ltd. + +SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial +Please see LICENSE in the repository root for full details. +*/ + +import { type Logger } from "matrix-js-sdk/lib/logger"; +import { + type CallMembership, + getRemoteLivekitPublishCandidates, + type LivekitPublishCandidate, + type Transport, +} from "matrix-js-sdk/lib/matrixrtc"; +import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager"; + +import { isCsApiUnavailable } from "./auth"; +import * as csApi from "./auth/csApi"; +import { decodeSFUJWTPayload } from "./auth/types"; + +export interface RemotePublishTransportParams { + client: csApi.ClientGetTokenParts; + /** The members whose SFUs are the candidates. */ + memberships: CallMembership[]; + ownMembershipIdentity: CallMembershipIdentityParts; + roomId: string; + slotId: string; + logger: Logger; +} + +/** + * The SFU of another member that its homeserver lets us publish on, as the + * transport to advertise, or null where no member's SFU will do. Candidates + * are tried oldest membership first, and each one is checked by asking for a + * publisher token through our homeserver (MSC4195): a homeserver may restrict + * remote publishers, which shows as a token without the publish grant. + */ +export async function findRemotePublishTransport({ + client, + memberships, + ownMembershipIdentity, + roomId, + slotId, + logger, +}: RemotePublishTransportParams): Promise { + logger.info( + "No local transport available, looking for another member's SFU to publish on", + ); + const candidates = getRemoteLivekitPublishCandidates( + memberships, + ownMembershipIdentity, + ); + for (const candidate of candidates) { + const outcome = await tryCandidate(candidate); + if (outcome === "usable") return candidate.transport; + if (outcome === "noCsApi") return null; + } + return null; + + /** "noCsApi" ends the search: without the endpoint no candidate can work. */ + async function tryCandidate({ + url, + serverName, + }: LivekitPublishCandidate): Promise<"usable" | "unusable" | "noCsApi"> { + try { + const { jwt } = await csApi.getSFUConfig({ + client, + membership: ownMembershipIdentity, + url, + serverName, + roomId, + slotId, + }); + if (decodeSFUJWTPayload(jwt).video.canPublish) { + logger.info(`Publishing on remote SFU ${url} of ${serverName}`); + return "usable"; + } + logger.info(`${serverName} does not let us publish on ${url}, skipping`); + } catch (e) { + if (isCsApiUnavailable(e)) { + logger.warn( + `Cannot use the MSC4195 get_token endpoint (${e}), so no remote SFU can be used`, + ); + return "noCsApi"; + } + logger.warn(`Failed to get a token for ${url} of ${serverName}:`, e); + } + return "unusable"; + } +} diff --git a/sdk/src/media-backend/livekit/auth/index.ts b/sdk/src/media-backend/livekit/auth/index.ts index a9756cf5e..409917fae 100644 --- a/sdk/src/media-backend/livekit/auth/index.ts +++ b/sdk/src/media-backend/livekit/auth/index.ts @@ -48,7 +48,7 @@ function isEndpointUnsupported(e: unknown): boolean { * (MSC4533), which the widget API reports as a plain error. A connection * error is neither: the endpoint may well exist. */ -function isCsApiUnavailable(e: unknown): boolean { +export function isCsApiUnavailable(e: unknown): boolean { if (isEndpointUnsupported(e)) return true; return !(e instanceof MatrixError) && !(e instanceof ConnectionError); } diff --git a/sdk/src/media-backend/livekit/auth/types.ts b/sdk/src/media-backend/livekit/auth/types.ts index 9c198f2d2..75e96321c 100644 --- a/sdk/src/media-backend/livekit/auth/types.ts +++ b/sdk/src/media-backend/livekit/auth/types.ts @@ -34,11 +34,13 @@ export interface SFUConfig { } /** The parts of the JWT payload the backend reads. */ -interface SFUJWTPayload { +export interface SFUJWTPayload { /** The subject: the participant identity the token carries. */ sub: string; video: { room: string; + /** Whether the token lets us publish; a homeserver may deny this to remote publishers. */ + canPublish?: boolean; }; } @@ -47,8 +49,7 @@ export function extractFullConfigFromToken(sfuConfig: { url: string; jwt: string; }): SFUConfig { - const [, payloadStr] = sfuConfig.jwt.split("."); - const payload = JSON.parse(global.atob(payloadStr)) as SFUJWTPayload; + const payload = decodeSFUJWTPayload(sfuConfig.jwt); return { jwt: sfuConfig.jwt, url: sfuConfig.url, @@ -56,3 +57,9 @@ export function extractFullConfigFromToken(sfuConfig: { livekitIdentity: payload.sub, }; } + +/** Decodes the payload of an SFU JWT, without verifying its signature. */ +export function decodeSFUJWTPayload(jwt: string): SFUJWTPayload { + const [, payloadStr] = jwt.split("."); + return JSON.parse(global.atob(payloadStr)) as SFUJWTPayload; +} diff --git a/sdk/src/utils/serverName.ts b/sdk/src/utils/serverName.ts index d6b6d685a..11a99b768 100644 --- a/sdk/src/utils/serverName.ts +++ b/sdk/src/utils/serverName.ts @@ -5,7 +5,23 @@ SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial Please see LICENSE in the repository root for full details. */ +import { type Transport } from "matrix-js-sdk/lib/matrixrtc"; + /** The homeserver a Matrix user id belongs to: everything after the first colon. */ export function serverNameOf(userId: string): string { return userId.replace(/^.*?:/, ""); } + +/** + * The homeserver that operates a transport: the one the transport names, else + * the one of the member advertising it. A member publishing on another + * homeserver's SFU (MSC4195) names that homeserver in `server_name`. + */ +export function serverNameOfTransport( + transport: Transport, + userId: string, +): string { + return typeof transport.server_name === "string" + ? transport.server_name + : serverNameOf(userId); +} diff --git a/src/room/__snapshots__/GroupCallErrorBoundary.test.tsx.snap b/src/room/__snapshots__/GroupCallErrorBoundary.test.tsx.snap index 7a1fc0457..cc7808a98 100644 --- a/src/room/__snapshots__/GroupCallErrorBoundary.test.tsx.snap +++ b/src/room/__snapshots__/GroupCallErrorBoundary.test.tsx.snap @@ -1228,7 +1228,7 @@ exports[`should have a close button when the host can dismiss us 1`] = ` Call is not supported

- The server is not configured to work with Element Call. Please contact your server admin (Domain: example.com, Error Code: MISSING_MATRIX_RTC_TRANSPORT). + The server is not configured to work with Element Call and no other call participant provides a usable transport. Please contact your server admin (Domain: example.com, Error Code: MISSING_MATRIX_RTC_TRANSPORT).