From 3b763d87f244868734788201b5375d9af298742d Mon Sep 17 00:00:00 2001 From: "Timo K." Date: Wed, 14 Oct 2026 16:40:27 +0200 Subject: [PATCH] make the transport discovery lk edgecase a EC exclusive workaround Dont add it to the sdk. --- package.json | 2 +- pnpm-lock.yaml | 10 +- sdk/SdkArchitecture.md | 18 ++-- sdk/src/RTCParticipation.test.ts | 57 +++++++---- sdk/src/RTCParticipation.ts | 25 ++++- sdk/src/matrixrtc/LocalTransport.test.ts | 57 +++++------ sdk/src/matrixrtc/LocalTransport.ts | 17 ++-- sdk/src/media-backend/api.ts | 16 +--- .../livekit/LivekitBackend.test.ts | 30 ------ .../media-backend/livekit/LivekitBackend.ts | 15 --- .../livekit/RemotePublishTransport.test.ts | 65 +++++++++---- .../livekit/RemotePublishTransport.ts | 95 ++++++++++--------- 12 files changed, 206 insertions(+), 201 deletions(-) diff --git a/package.json b/package.json index a670d84d4..f9823d4f5 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#9405287c22797e9d69acd366c47bf412fa4384c8", + "matrix-js-sdk": "github:matrix-org/matrix-js-sdk#3f05dc4d104a3dfc753c5e3549a18c74fab4b076", "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 ecafc61f5..f12307a69 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#9405287c22797e9d69acd366c47bf412fa4384c8 - version: https://codeload.github.com/matrix-org/matrix-js-sdk/tar.gz/9405287c22797e9d69acd366c47bf412fa4384c8 + specifier: github:matrix-org/matrix-js-sdk#3f05dc4d104a3dfc753c5e3549a18c74fab4b076 + version: https://codeload.github.com/matrix-org/matrix-js-sdk/tar.gz/3f05dc4d104a3dfc753c5e3549a18c74fab4b076 matrix-widget-api: specifier: ^1.18.0 version: 1.20.0 @@ -4673,8 +4673,8 @@ 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/9405287c22797e9d69acd366c47bf412fa4384c8: - resolution: {gitHosted: true, integrity: sha512-pYk38ggTEdM0WWXLDiFoKmyRn3wo3o8m1ge0iCtxjBmgAW5qoQ2Ifd8oHVy+gu6PQe/CfLuKPXEXH8hff9C8GA==, tarball: https://codeload.github.com/matrix-org/matrix-js-sdk/tar.gz/9405287c22797e9d69acd366c47bf412fa4384c8} + matrix-js-sdk@https://codeload.github.com/matrix-org/matrix-js-sdk/tar.gz/3f05dc4d104a3dfc753c5e3549a18c74fab4b076: + resolution: {gitHosted: true, integrity: sha512-Ly8eS7fNNKDo9KmPvTkE49KvD8iMUmrVWJMNERNuxc8UksMLMG36NLz3fs8HytdwrB7AKPfyVo392VZsxILDIg==, tarball: https://codeload.github.com/matrix-org/matrix-js-sdk/tar.gz/3f05dc4d104a3dfc753c5e3549a18c74fab4b076} version: 44.0.0 engines: {node: '>=22.0.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/9405287c22797e9d69acd366c47bf412fa4384c8: + matrix-js-sdk@https://codeload.github.com/matrix-org/matrix-js-sdk/tar.gz/3f05dc4d104a3dfc753c5e3549a18c74fab4b076: 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 4a184f6eb..19b70d3ff 100644 --- a/sdk/SdkArchitecture.md +++ b/sdk/SdkArchitecture.md @@ -101,8 +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) ─▶│ │ + │ │ (none, LiveKit only: a member's SFU │ │ + │ │ that lets us publish, get_token) ─▶│ │ │ │ prepareLocalTransport (backend): │ │ │ │ publisher token: CS API get_token, │ │ │ │ or OpenID token then /sfu/get ─────▶│ │ @@ -539,10 +539,6 @@ 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, @@ -562,9 +558,10 @@ 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), asking the -backend for another member's transport only where that list and the host's -fallback are empty, the membership +homeserver's list (a homeserver endpoint, so not a backend's) with, for the +LiveKit backend only, a last resort of another member's SFU that lets us +publish (`findRemotePublishLivekitTransport`, wired in by `RTCParticipation` rather +than by the backend interface, since it is an MSC4195 matter), 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 @@ -574,8 +571,7 @@ 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$` and the one membership -snapshot `findRemotePublishTransport` is handed, registering each +and learns of remote members only through `mediaFor$`, 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 48e63fbcc..fc01be5d8 100644 --- a/sdk/src/RTCParticipation.test.ts +++ b/sdk/src/RTCParticipation.test.ts @@ -8,6 +8,7 @@ Please see LICENSE in the repository root for full details. import { type MatrixClient, type Room, SyncState } from "matrix-js-sdk"; import { logger } from "matrix-js-sdk/lib/logger"; import { + type CallMembership, MatrixRTCSessionEvent, MembershipManagerEvent, type Transport, @@ -38,13 +39,26 @@ import { constant } from "./reactive/Behavior"; import { ObservableScope } from "./reactive/ObservableScope"; import { createRTCParticipation } from "./RTCParticipation"; import { transportId } from "./matrixrtc/Transports"; -import { flushPromises, mockEmitter, MockRTCSession } from "./utils/test"; +import { + flushPromises, + mockEmitter, + mockRtcMembership, + MockRTCSession, +} from "./utils/test"; const transport: Transport = { type: "livekit", livekit_service_url: "https://lk.example.org", }; const userId = "@alice:example.org"; +/** A token with the publish grant, as a homeserver issues it for its own SFU. */ +const publisherJWT = [ + {}, + { sub: "identity", video: { room: "room", canPublish: true } }, + {}, +] + .map((d) => global.btoa(JSON.stringify(d))) + .join("."); const deviceId = "AAAA"; const legacyMemberId = `${userId}:${deviceId}`; @@ -84,7 +98,6 @@ 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>; } @@ -92,7 +105,13 @@ interface FakeBackend { function setup( options: Partial = {}, slotOptions: { matrixRTCMode?: MatrixRTCMode } = {}, - homeserverTransports: Transport[] = [transport], + { + homeserverTransports = [transport], + memberships = [], + }: { + homeserverTransports?: Transport[]; + memberships?: CallMembership[]; + } = {}, ): { scope: ObservableScope; participation: ReturnType["participation"]; @@ -105,7 +124,7 @@ function setup( } { const scope = new ObservableScope(); const room = { roomId: "!room:example.org" } as Room; - const session = new MockRTCSession(room); + const session = new MockRTCSession(room, memberships); const mockedSession = session.asMockedSession(); const join = vi.spyOn(session, "join"); const client = { @@ -115,6 +134,7 @@ function setup( getDomain: () => "example.org", getSyncState: () => SyncState.Syncing, _unstable_getRTCTransports: vi.fn().mockResolvedValue(homeserverTransports), + _unstable_getLivekitToken: vi.fn().mockResolvedValue({ jwt: publisherJWT }), } as unknown as MatrixClient; const members$ = new BehaviorSubject([]); @@ -126,7 +146,6 @@ 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), @@ -139,7 +158,6 @@ function setup( ): MediaBackend => { const backend: MediaBackend = { transportType: "livekit", - findRemotePublishTransport: fake.findRemotePublishTransport, prepareLocalTransport: async () => fake.prepared.promise, delegateDelayedLeave: fake.delegateDelayedLeave, local: { @@ -225,22 +243,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); + const bobOnSfu = mockRtcMembership("@bob:example.org", "BOB", { + fociPreferred: [{ type: "livekit", url: "wss://sfu.example.org" }], + }); + const { join, fake } = setup( + {}, + {}, + { homeserverTransports: [], memberships: [bobOnSfu] }, + ); fake.prepared.resolve({ canDelegateDelayedLeave: false }); await flushPromises(); + await flushPromises(); - expect(fake.findRemotePublishTransport).toHaveBeenCalledWith( - session.memberships, - ); expect(join).toHaveBeenCalledWith( expect.anything(), - [bobsTransport], + [ + { + type: "livekit", + url: "wss://sfu.example.org", + server_name: "example.org", + }, + ], expect.anything(), ); }); diff --git a/sdk/src/RTCParticipation.ts b/sdk/src/RTCParticipation.ts index 31a39240a..5f6a81796 100644 --- a/sdk/src/RTCParticipation.ts +++ b/sdk/src/RTCParticipation.ts @@ -11,6 +11,7 @@ import { type MatrixRTCSession as JsSdkRTCSession, MatrixRTCSessionEvent, MembershipManagerEvent, + type Transport, } from "matrix-js-sdk/lib/matrixrtc"; import { type IMembershipManager } from "matrix-js-sdk/lib/matrixrtc/IMembershipManager"; import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager"; @@ -50,6 +51,7 @@ import { type MediaKey, } from "./media-backend/api"; import { createLivekitBackend } from "./media-backend/livekit/LivekitBackend"; +import { findRemotePublishLivekitTransport } from "./media-backend/livekit/RemotePublishTransport"; import { createHomeserverConnected$ } from "./matrixrtc/HomeserverConnected"; import { joinJsSdkSession } from "./matrixrtc/joinJsSdkSession"; import { @@ -130,14 +132,29 @@ export function createRTCParticipation( logger, }); + // If the backend is a LiveKit: We check it for transports as well. (as a fallbacke) + // The livekit backend is a source of transports too. + // (MSC4195). LiveKit only, so it stays out of the backend interface + const transportsFromBackends = async (): Promise => + backend.transportType === "livekit" + ? findRemotePublishLivekitTransport({ + client, + // 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 + memberships: jsSdkSession.memberships, + ownMembershipIdentity, + roomId: room.roomId, + slotId: jsSdkSession.slotId, + logger, + }) + : []; + const preparedTransport$ = from( 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, + backend.transportType, options, + transportsFromBackends, logger, ).then(async (transport): Promise => ({ transport, diff --git a/sdk/src/matrixrtc/LocalTransport.test.ts b/sdk/src/matrixrtc/LocalTransport.test.ts index a2c834372..b81958e2e 100644 --- a/sdk/src/matrixrtc/LocalTransport.test.ts +++ b/sdk/src/matrixrtc/LocalTransport.test.ts @@ -10,7 +10,6 @@ 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 = { @@ -26,7 +25,11 @@ const bobsTransport: Transport = { url: "wss://sfu.bob.com", server_name: "bob.com", }; -const bob = mockRtcMembership("@bob:bob.com", "BOB"); +const carolsTransport: Transport = { + type: "livekit", + url: "wss://sfu.carol.com", + server_name: "carol.com", +}; function client( advertised: Transport[], @@ -37,13 +40,10 @@ function client( }; } -function backend( - remote: Transport | null = null, -): Parameters[1] { - return { - transportType: "livekit", - findRemotePublishTransport: vi.fn().mockResolvedValue(remote), - }; +function fromBackends( + transports: Transport[] = [], +): ReturnType Promise>> { + return vi.fn(async () => Promise.resolve(transports)); } describe("discoverLocalTransport", () => { @@ -53,9 +53,9 @@ describe("discoverLocalTransport", () => { await expect( discoverLocalTransport( c, - backend(), - [], + "livekit", { transport: hostTransport }, + fromBackends(), logger, ), ).resolves.toBe(hostTransport); @@ -66,9 +66,9 @@ describe("discoverLocalTransport", () => { await expect( discoverLocalTransport( client([homeserverTransport]), - backend(), - [], + "livekit", { fallbackTransport: hostTransport }, + fromBackends(), logger, ), ).resolves.toEqual(homeserverTransport); @@ -78,44 +78,47 @@ describe("discoverLocalTransport", () => { await expect( discoverLocalTransport( client([{ type: "not_livekit" }]), - backend(), - [], + "livekit", { fallbackTransport: hostTransport }, + fromBackends(), logger, ), ).resolves.toEqual(hostTransport); }); - it("prefers a local transport over other members' SFUs", async () => { - const b = backend(bobsTransport); + it("prefers a local transport over the backends' without asking them", async () => { + const backends = fromBackends([bobsTransport]); await expect( discoverLocalTransport( client([homeserverTransport]), - b, - [bob], + "livekit", {}, + backends, logger, ), ).resolves.toEqual(homeserverTransport); - expect(b.findRemotePublishTransport).not.toHaveBeenCalled(); + expect(backends).not.toHaveBeenCalled(); }); - it("publishes on another member's SFU where there is no local transport", async () => { - const b = backend(bobsTransport); - + it("takes the first transport the backends provide where there is no local one", async () => { await expect( - discoverLocalTransport(client([]), b, [bob], {}, logger), + discoverLocalTransport( + client([]), + "livekit", + {}, + fromBackends([bobsTransport, carolsTransport]), + 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([]), - backend(), - [bob], + "livekit", {}, + fromBackends(), logger, ); diff --git a/sdk/src/matrixrtc/LocalTransport.ts b/sdk/src/matrixrtc/LocalTransport.ts index 09ffd1f9b..4c967ce39 100644 --- a/sdk/src/matrixrtc/LocalTransport.ts +++ b/sdk/src/matrixrtc/LocalTransport.ts @@ -7,42 +7,39 @@ 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, - type Transport, -} from "matrix-js-sdk/lib/matrixrtc"; +import { 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, - * else another member's that the backend can publish on. + * else the first the backends can provide, which are asked only then. * * @throws {MatrixRTCTransportMissingError} where none of these exists. */ export async function discoverLocalTransport( client: Pick, - backend: Pick, - memberships: CallMembership[], + transportType: string, { transport: override, fallbackTransport, }: Pick, + transportsFromBackends: () => Promise, logger: Logger, ): Promise { const transport = override ?? (await new TransportAutoDiscovery({ client, - transportType: backend.transportType, + transportType, fallbackTransport, logger, }).discoverPreferredTransport()) ?? - (await backend.findRemotePublishTransport(memberships)); + (await transportsFromBackends()).at(0) ?? + null; 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 b80277d00..4d5f5562f 100644 --- a/sdk/src/media-backend/api.ts +++ b/sdk/src/media-backend/api.ts @@ -13,10 +13,7 @@ Please see LICENSE in the repository root for full details. */ import { type Logger } from "matrix-js-sdk/lib/logger"; -import { - type CallMembership, - type Transport, -} from "matrix-js-sdk/lib/matrixrtc"; +import { type Transport } from "matrix-js-sdk/lib/matrixrtc"; import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager"; import { type Observable } from "rxjs"; @@ -119,17 +116,6 @@ 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 ddf60fd52..193fe3c8c 100644 --- a/sdk/src/media-backend/livekit/LivekitBackend.test.ts +++ b/sdk/src/media-backend/livekit/LivekitBackend.test.ts @@ -25,7 +25,6 @@ import { defaultSessionTimings } from "../../config"; import { flushPromises, mockEmitter, - mockRtcMembership, ownMemberMock, testScope, } from "../../utils/test"; @@ -276,35 +275,6 @@ describe("createLivekitBackend", () => { 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); diff --git a/sdk/src/media-backend/livekit/LivekitBackend.ts b/sdk/src/media-backend/livekit/LivekitBackend.ts index c139f6f52..f2a04720d 100644 --- a/sdk/src/media-backend/livekit/LivekitBackend.ts +++ b/sdk/src/media-backend/livekit/LivekitBackend.ts @@ -8,7 +8,6 @@ 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"; @@ -63,7 +62,6 @@ import { createDataChannel$ } from "./DataChannel"; import { createKeyProvider } from "./KeyProvider"; import { createLivekitMemberMedia } from "./LivekitMemberMedia"; import { createLivekitLocalMedia } from "./LocalMedia"; -import { findRemotePublishTransport } from "./RemotePublishTransport"; export interface LivekitBackendOptions { /** @@ -150,18 +148,6 @@ export function createLivekitBackend( const ownServerName = serverNameOf(ownMembershipIdentity.userId); - const remotePublishTransport = async ( - memberships: CallMembership[], - ): Promise => - findRemotePublishTransport({ - client, - memberships, - ownMembershipIdentity, - roomId, - slotId, - logger, - }); - const prepareLocalTransport = async ( transport: Transport, ): Promise => { @@ -286,7 +272,6 @@ 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 index 455366af6..99e16b6c3 100644 --- a/sdk/src/media-backend/livekit/RemotePublishTransport.test.ts +++ b/sdk/src/media-backend/livekit/RemotePublishTransport.test.ts @@ -12,7 +12,7 @@ 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"; +import { findRemotePublishLivekitTransport } from "./RemotePublishTransport"; /** A JWT whose payload states whether the issuing server lets us publish. */ function jwtWithPublishGrant(canPublish: boolean): string { @@ -27,6 +27,8 @@ function jwtWithPublishGrant(canPublish: boolean): string { const bobSfu = { type: "livekit", url: "wss://sfu.bob.com" }; const carolSfu = { type: "livekit", url: "wss://sfu.carol.com" }; +const bobsTransport = { ...bobSfu, server_name: "bob.com" }; +const carolsTransport = { ...carolSfu, server_name: "carol.com" }; // Bob is the oldest member, Carol joined after him const bob = mockRtcMembership("@bob:bob.com", "BOB", { @@ -37,24 +39,41 @@ const carol = mockRtcMembership("@carol:carol.com", "CAROL", { fociPreferred: [carolSfu], membership: { created_ts: 200 }, }); -const legacyDave = mockRtcMembership("@dave:dave.com", "DAVE", { +// Bob's other device, and Dave who publishes on Bob's SFU remotely +const bobsLaptop = mockRtcMembership("@bob:bob.com", "LAPTOP", { + fociPreferred: [bobSfu], + membership: { created_ts: 150 }, +}); +const dave = mockRtcMembership("@dave:dave.com", "DAVE", { + fociPreferred: [bobsTransport], + membership: { created_ts: 160 }, +}); +const legacyErin = mockRtcMembership("@erin:erin.com", "ERIN", { fociPreferred: [ - { type: "livekit", livekit_service_url: "https://jwt.dave.com" }, + { type: "livekit", livekit_service_url: "https://jwt.erin.com" }, ], membership: { created_ts: 50 }, }); +const ourselves = mockRtcMembership( + ownMemberMock.userId, + ownMemberMock.deviceId, + { + fociPreferred: [{ type: "livekit", url: "wss://sfu.example.org" }], + membership: { created_ts: 10 }, + }, +); const csApiMissing = new MatrixError( { errcode: "M_UNRECOGNIZED", error: "" }, 404, ); -describe("findRemotePublishTransport", () => { +describe("findRemotePublishLivekitTransport", () => { let client: Mocked; const find = async ( memberships: CallMembership[], - ): ReturnType => - findRemotePublishTransport({ + ): ReturnType => + findRemotePublishLivekitTransport({ client, memberships, ownMembershipIdentity: ownMemberMock, @@ -72,11 +91,7 @@ describe("findRemotePublishTransport", () => { }); 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", - }); + await expect(find([carol, bob])).resolves.toEqual(bobsTransport); expect(client._unstable_getLivekitToken).toHaveBeenCalledTimes(1); expect(client._unstable_getLivekitToken).toHaveBeenCalledWith({ url: "wss://sfu.bob.com", @@ -87,10 +102,22 @@ describe("findRemotePublishTransport", () => { }); }); - it("ignores members publishing through a legacy JWT service", async () => { - await expect(find([legacyDave, bob])).resolves.toMatchObject({ - server_name: "bob.com", + it("asks about an SFU once however many members are on it", async () => { + client._unstable_getLivekitToken.mockResolvedValue({ + jwt: jwtWithPublishGrant(false), }); + + await expect(find([dave, bobsLaptop, bob])).resolves.toBeNull(); + expect(client._unstable_getLivekitToken).toHaveBeenCalledTimes(1); + }); + + it("skips our own membership", async () => { + await expect(find([ourselves, bob])).resolves.toEqual(bobsTransport); + expect(client._unstable_getLivekitToken).toHaveBeenCalledTimes(1); + }); + + it("ignores members publishing through a legacy JWT service", async () => { + await expect(find([legacyErin, bob])).resolves.toEqual(bobsTransport); expect(client._unstable_getLivekitToken).toHaveBeenCalledTimes(1); }); @@ -102,9 +129,7 @@ describe("findRemotePublishTransport", () => { }), ); - await expect(find([bob, carol])).resolves.toMatchObject({ - server_name: "carol.com", - }); + await expect(find([bob, carol])).resolves.toEqual(carolsTransport); expect(client._unstable_getLivekitToken).toHaveBeenCalledTimes(2); }); @@ -121,9 +146,7 @@ describe("findRemotePublishTransport", () => { : Promise.resolve({ jwt: jwtWithPublishGrant(true) }), ); - await expect(find([bob, carol])).resolves.toMatchObject({ - server_name: "carol.com", - }); + await expect(find([bob, carol])).resolves.toEqual(carolsTransport); }); it("finds nothing once all candidates are exhausted", async () => { @@ -143,7 +166,7 @@ describe("findRemotePublishTransport", () => { }); it("finds nothing where there are no other members", async () => { - await expect(find([])).resolves.toBeNull(); + await expect(find([ourselves])).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 index 988a6ed3d..3efce5ad7 100644 --- a/sdk/src/media-backend/livekit/RemotePublishTransport.ts +++ b/sdk/src/media-backend/livekit/RemotePublishTransport.ts @@ -8,8 +8,7 @@ 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, + isUnstableLivekitTransport, type Transport, } from "matrix-js-sdk/lib/matrixrtc"; import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager"; @@ -30,15 +29,15 @@ export interface RemotePublishTransportParams { /** * 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 + * transport to advertise, or null where none does. Candidates are tried + * oldest membership first, each SFU once, and each 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({ +export async function findRemotePublishLivekitTransport({ client, memberships, - ownMembershipIdentity, + ownMembershipIdentity: own, roomId, slotId, logger, @@ -46,45 +45,51 @@ export async function findRemotePublishTransport({ 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; + const tried = new Set(); + const byAge = [...memberships].sort((a, b) => a.createdTs() - b.createdTs()); + for (const membership of byAge) { + if ( + membership.userId === own.userId && + membership.deviceId === own.deviceId + ) + continue; + for (const transport of membership.transports) { + // Publishing remotely goes through the client-server API, which a JWT + // service cannot stand in for, so only a `url` qualifies + if (!isUnstableLivekitTransport(transport) || !("url" in transport)) + continue; + const { url } = transport; + const serverName = membership.getTransportServerName(transport); + const key = JSON.stringify([url, serverName]); + if (tried.has(key)) continue; + tried.add(key); + try { + const { jwt } = await csApi.getSFUConfig({ + client, + membership: own, + url, + serverName, + roomId, + slotId, + }); + if (decodeSFUJWTPayload(jwt).video.canPublish) { + logger.info(`Publishing on remote SFU ${url} of ${serverName}`); + return { type: "livekit", url, server_name: serverName }; + } + logger.info( + `${serverName} does not let us publish on ${url}, skipping`, + ); + } catch (e) { + if (isCsApiUnavailable(e)) { + // Without the endpoint no further candidate can work + logger.warn( + `Cannot use the MSC4195 get_token endpoint (${e}), so no remote SFU can be used`, + ); + return null; + } + logger.warn(`Failed to get a token for ${url} of ${serverName}:`, e); + } + } } 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"; - } }