make the transport discovery lk edgecase a EC exclusive workaround

Dont add it to the sdk.
This commit is contained in:
Timo K.
2026-10-14 16:40:27 +02:00
parent 8638d4f159
commit 3b763d87f2
12 changed files with 206 additions and 201 deletions
+1 -1
View File
@@ -107,7 +107,7 @@
"livekit-client": "^2.18.1", "livekit-client": "^2.18.1",
"lodash-es": "^4.17.21", "lodash-es": "^4.17.21",
"loglevel": "^1.9.1", "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", "matrix-widget-api": "^1.18.0",
"node-stdlib-browser": "^1.3.1", "node-stdlib-browser": "^1.3.1",
"normalize.css": "^8.0.1", "normalize.css": "^8.0.1",
+5 -5
View File
@@ -191,8 +191,8 @@ importers:
specifier: ^1.9.1 specifier: ^1.9.1
version: 1.9.2 version: 1.9.2
matrix-js-sdk: matrix-js-sdk:
specifier: github:matrix-org/matrix-js-sdk#9405287c22797e9d69acd366c47bf412fa4384c8 specifier: github:matrix-org/matrix-js-sdk#3f05dc4d104a3dfc753c5e3549a18c74fab4b076
version: https://codeload.github.com/matrix-org/matrix-js-sdk/tar.gz/9405287c22797e9d69acd366c47bf412fa4384c8 version: https://codeload.github.com/matrix-org/matrix-js-sdk/tar.gz/3f05dc4d104a3dfc753c5e3549a18c74fab4b076
matrix-widget-api: matrix-widget-api:
specifier: ^1.18.0 specifier: ^1.18.0
version: 1.20.0 version: 1.20.0
@@ -4673,8 +4673,8 @@ packages:
matrix-events-sdk@0.0.1: matrix-events-sdk@0.0.1:
resolution: {integrity: sha512-1QEOsXO+bhyCroIe2/A5OwaxHvBm7EsSQ46DEDn8RBIfQwN5HWBpFvyWWR4QY0KHPPnnJdI99wgRiAl7Ad5qaA==} resolution: {integrity: sha512-1QEOsXO+bhyCroIe2/A5OwaxHvBm7EsSQ46DEDn8RBIfQwN5HWBpFvyWWR4QY0KHPPnnJdI99wgRiAl7Ad5qaA==}
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:
resolution: {gitHosted: true, integrity: sha512-pYk38ggTEdM0WWXLDiFoKmyRn3wo3o8m1ge0iCtxjBmgAW5qoQ2Ifd8oHVy+gu6PQe/CfLuKPXEXH8hff9C8GA==, tarball: https://codeload.github.com/matrix-org/matrix-js-sdk/tar.gz/9405287c22797e9d69acd366c47bf412fa4384c8} resolution: {gitHosted: true, integrity: sha512-Ly8eS7fNNKDo9KmPvTkE49KvD8iMUmrVWJMNERNuxc8UksMLMG36NLz3fs8HytdwrB7AKPfyVo392VZsxILDIg==, tarball: https://codeload.github.com/matrix-org/matrix-js-sdk/tar.gz/3f05dc4d104a3dfc753c5e3549a18c74fab4b076}
version: 44.0.0 version: 44.0.0
engines: {node: '>=22.0.0'} engines: {node: '>=22.0.0'}
@@ -10227,7 +10227,7 @@ snapshots:
matrix-events-sdk@0.0.1: {} 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: dependencies:
'@babel/runtime': 8.0.0 '@babel/runtime': 8.0.0
'@matrix-org/matrix-sdk-crypto-wasm': 18.9.0 '@matrix-org/matrix-sdk-crypto-wasm': 18.9.0
+7 -11
View File
@@ -101,8 +101,8 @@ LiveKit participants.
│ create ───▶│ │ │ │ create ───▶│ │ │
│ │ GET rtc/transports (session) ────────▶│ │ │ │ GET rtc/transports (session) ────────▶│ │
│ │◀── first transport of the backend's type │ │ │◀── first transport of the backend's type │
│ │ (none: findRemotePublishTransport, │ │ │ │ (none, LiveKit only: a member's SFU │ │
│ │ a member's SFU we may publish on) ─▶│ │ │ │ that lets us publish, get_token) ─▶│ │
│ │ prepareLocalTransport (backend): │ │ │ │ prepareLocalTransport (backend): │ │
│ │ publisher token: CS API get_token, │ │ │ │ publisher token: CS API get_token, │ │
│ │ or OpenID token then /sfu/get ─────▶│ │ │ │ or OpenID token then /sfu/get ─────▶│ │
@@ -539,10 +539,6 @@ type MemberMediaFields = Pick<RTCMember, "tracks$" | "encryptionError$">;
export interface MediaBackend { export interface MediaBackend {
readonly transportType: string; // "livekit" readonly transportType: string; // "livekit"
/** Another member's transport we may publish on, where the client found none. */
findRemotePublishTransport(
memberships: CallMembership[],
): Promise<Transport | null>;
/** Authenticate with the transport the client found; the probe for delegation runs here. */ /** Authenticate with the transport the client found; the probe for delegation runs here. */
prepareLocalTransport( prepareLocalTransport(
transport: Transport, transport: Transport,
@@ -562,9 +558,10 @@ export interface MediaBackend {
``` ```
What stays on the client's side: discovery of the local transport from the 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 homeserver's list (a homeserver endpoint, so not a backend's) with, for the
backend for another member's transport only where that list and the host's LiveKit backend only, a last resort of another member's SFU that lets us
fallback are empty, the membership 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 state machine in `LocalMembership.ts` and its combination with
`local.connectionState$` into `state$`, the choice of delayed leave timings, `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 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 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 that needs the client takes a `Pick` of exactly what it calls in its own options
(`createLivekitBackend(scope, context, { client, mediaQuality })`), (`createLivekitBackend(scope, context, { client, mediaQuality })`),
and learns of remote members only through `mediaFor$` and the one membership and learns of remote members only through `mediaFor$`, registering each
snapshot `findRemotePublishTransport` is handed, registering each
membership's transport for as long as the member's scope lives. Media keys reach 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 it as a stream, `mediaKeys$`, replayed on subscribe; the js-sdk session never
does. A host supplies its own backend through `MatrixRTCClientOptions.backend`; does. A host supplies its own backend through `MatrixRTCClientOptions.backend`;
+40 -17
View File
@@ -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 { type MatrixClient, type Room, SyncState } from "matrix-js-sdk";
import { logger } from "matrix-js-sdk/lib/logger"; import { logger } from "matrix-js-sdk/lib/logger";
import { import {
type CallMembership,
MatrixRTCSessionEvent, MatrixRTCSessionEvent,
MembershipManagerEvent, MembershipManagerEvent,
type Transport, type Transport,
@@ -38,13 +39,26 @@ import { constant } from "./reactive/Behavior";
import { ObservableScope } from "./reactive/ObservableScope"; import { ObservableScope } from "./reactive/ObservableScope";
import { createRTCParticipation } from "./RTCParticipation"; import { createRTCParticipation } from "./RTCParticipation";
import { transportId } from "./matrixrtc/Transports"; import { transportId } from "./matrixrtc/Transports";
import { flushPromises, mockEmitter, MockRTCSession } from "./utils/test"; import {
flushPromises,
mockEmitter,
mockRtcMembership,
MockRTCSession,
} from "./utils/test";
const transport: Transport = { const transport: Transport = {
type: "livekit", type: "livekit",
livekit_service_url: "https://lk.example.org", livekit_service_url: "https://lk.example.org",
}; };
const userId = "@alice: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 deviceId = "AAAA";
const legacyMemberId = `${userId}:${deviceId}`; const legacyMemberId = `${userId}:${deviceId}`;
@@ -84,7 +98,6 @@ interface FakeBackend {
connectionState$: BehaviorSubject<MediaConnectionState | Error>; connectionState$: BehaviorSubject<MediaConnectionState | Error>;
data$: Subject<DataPacket>; data$: Subject<DataPacket>;
setPublishing: Mock<(publish: boolean) => void>; setPublishing: Mock<(publish: boolean) => void>;
findRemotePublishTransport: Mock<() => Promise<Transport | null>>;
delegateDelayedLeave: Mock<(delayId: string) => Promise<void>>; delegateDelayedLeave: Mock<(delayId: string) => Promise<void>>;
mediaFor: Mock<(scope: ObservableScope, member: RTCMember) => MemberMedia>; mediaFor: Mock<(scope: ObservableScope, member: RTCMember) => MemberMedia>;
} }
@@ -92,7 +105,13 @@ interface FakeBackend {
function setup( function setup(
options: Partial<RTCParticipationOptions> = {}, options: Partial<RTCParticipationOptions> = {},
slotOptions: { matrixRTCMode?: MatrixRTCMode } = {}, slotOptions: { matrixRTCMode?: MatrixRTCMode } = {},
homeserverTransports: Transport[] = [transport], {
homeserverTransports = [transport],
memberships = [],
}: {
homeserverTransports?: Transport[];
memberships?: CallMembership[];
} = {},
): { ): {
scope: ObservableScope; scope: ObservableScope;
participation: ReturnType<typeof createRTCParticipation>["participation"]; participation: ReturnType<typeof createRTCParticipation>["participation"];
@@ -105,7 +124,7 @@ function setup(
} { } {
const scope = new ObservableScope(); const scope = new ObservableScope();
const room = { roomId: "!room:example.org" } as Room; const room = { roomId: "!room:example.org" } as Room;
const session = new MockRTCSession(room); const session = new MockRTCSession(room, memberships);
const mockedSession = session.asMockedSession(); const mockedSession = session.asMockedSession();
const join = vi.spyOn(session, "join"); const join = vi.spyOn(session, "join");
const client = { const client = {
@@ -115,6 +134,7 @@ function setup(
getDomain: () => "example.org", getDomain: () => "example.org",
getSyncState: () => SyncState.Syncing, getSyncState: () => SyncState.Syncing,
_unstable_getRTCTransports: vi.fn().mockResolvedValue(homeserverTransports), _unstable_getRTCTransports: vi.fn().mockResolvedValue(homeserverTransports),
_unstable_getLivekitToken: vi.fn().mockResolvedValue({ jwt: publisherJWT }),
} as unknown as MatrixClient; } as unknown as MatrixClient;
const members$ = new BehaviorSubject<RTCMember[]>([]); const members$ = new BehaviorSubject<RTCMember[]>([]);
@@ -126,7 +146,6 @@ function setup(
), ),
data$: new Subject<DataPacket>(), data$: new Subject<DataPacket>(),
setPublishing: vi.fn(), setPublishing: vi.fn(),
findRemotePublishTransport: vi.fn(async () => Promise.resolve(null)),
delegateDelayedLeave: vi.fn(async () => Promise.resolve()), delegateDelayedLeave: vi.fn(async () => Promise.resolve()),
mediaFor: vi.fn(() => ({ mediaFor: vi.fn(() => ({
tracks$: constant(null), tracks$: constant(null),
@@ -139,7 +158,6 @@ function setup(
): MediaBackend => { ): MediaBackend => {
const backend: MediaBackend = { const backend: MediaBackend = {
transportType: "livekit", transportType: "livekit",
findRemotePublishTransport: fake.findRemotePublishTransport,
prepareLocalTransport: async () => fake.prepared.promise, prepareLocalTransport: async () => fake.prepared.promise,
delegateDelayedLeave: fake.delegateDelayedLeave, delegateDelayedLeave: fake.delegateDelayedLeave,
local: { local: {
@@ -225,22 +243,27 @@ describe("createRTCParticipation", () => {
}); });
it("publishes on another member's SFU where the homeserver advertises none", async () => { it("publishes on another member's SFU where the homeserver advertises none", async () => {
const bobsTransport: Transport = { const bobOnSfu = mockRtcMembership("@bob:example.org", "BOB", {
type: "livekit", fociPreferred: [{ type: "livekit", url: "wss://sfu.example.org" }],
url: "wss://sfu.bob.com", });
server_name: "bob.com", const { join, fake } = setup(
}; {},
const { session, join, fake } = setup({}, {}, []); {},
fake.findRemotePublishTransport.mockResolvedValue(bobsTransport); { homeserverTransports: [], memberships: [bobOnSfu] },
);
fake.prepared.resolve({ canDelegateDelayedLeave: false }); fake.prepared.resolve({ canDelegateDelayedLeave: false });
await flushPromises(); await flushPromises();
await flushPromises();
expect(fake.findRemotePublishTransport).toHaveBeenCalledWith(
session.memberships,
);
expect(join).toHaveBeenCalledWith( expect(join).toHaveBeenCalledWith(
expect.anything(), expect.anything(),
[bobsTransport], [
{
type: "livekit",
url: "wss://sfu.example.org",
server_name: "example.org",
},
],
expect.anything(), expect.anything(),
); );
}); });
+21 -4
View File
@@ -11,6 +11,7 @@ import {
type MatrixRTCSession as JsSdkRTCSession, type MatrixRTCSession as JsSdkRTCSession,
MatrixRTCSessionEvent, MatrixRTCSessionEvent,
MembershipManagerEvent, MembershipManagerEvent,
type Transport,
} from "matrix-js-sdk/lib/matrixrtc"; } from "matrix-js-sdk/lib/matrixrtc";
import { type IMembershipManager } from "matrix-js-sdk/lib/matrixrtc/IMembershipManager"; import { type IMembershipManager } from "matrix-js-sdk/lib/matrixrtc/IMembershipManager";
import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager"; import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager";
@@ -50,6 +51,7 @@ import {
type MediaKey, type MediaKey,
} from "./media-backend/api"; } from "./media-backend/api";
import { createLivekitBackend } from "./media-backend/livekit/LivekitBackend"; import { createLivekitBackend } from "./media-backend/livekit/LivekitBackend";
import { findRemotePublishLivekitTransport } from "./media-backend/livekit/RemotePublishTransport";
import { createHomeserverConnected$ } from "./matrixrtc/HomeserverConnected"; import { createHomeserverConnected$ } from "./matrixrtc/HomeserverConnected";
import { joinJsSdkSession } from "./matrixrtc/joinJsSdkSession"; import { joinJsSdkSession } from "./matrixrtc/joinJsSdkSession";
import { import {
@@ -130,14 +132,29 @@ export function createRTCParticipation(
logger, 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<Transport[]> =>
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( const preparedTransport$ = from(
discoverLocalTransport( discoverLocalTransport(
client, client,
backend, backend.transportType,
// 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, options,
transportsFromBackends,
logger, logger,
).then(async (transport): Promise<PreparedTransport> => ({ ).then(async (transport): Promise<PreparedTransport> => ({
transport, transport,
+30 -27
View File
@@ -10,7 +10,6 @@ import { type Transport } from "matrix-js-sdk/lib/matrixrtc";
import { describe, expect, it, vi } from "vitest"; import { describe, expect, it, vi } from "vitest";
import { MatrixRTCTransportMissingError } from "../errors"; import { MatrixRTCTransportMissingError } from "../errors";
import { mockRtcMembership } from "../utils/test";
import { discoverLocalTransport } from "./LocalTransport"; import { discoverLocalTransport } from "./LocalTransport";
const homeserverTransport: Transport = { const homeserverTransport: Transport = {
@@ -26,7 +25,11 @@ const bobsTransport: Transport = {
url: "wss://sfu.bob.com", url: "wss://sfu.bob.com",
server_name: "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( function client(
advertised: Transport[], advertised: Transport[],
@@ -37,13 +40,10 @@ function client(
}; };
} }
function backend( function fromBackends(
remote: Transport | null = null, transports: Transport[] = [],
): Parameters<typeof discoverLocalTransport>[1] { ): ReturnType<typeof vi.fn<() => Promise<Transport[]>>> {
return { return vi.fn(async () => Promise.resolve(transports));
transportType: "livekit",
findRemotePublishTransport: vi.fn().mockResolvedValue(remote),
};
} }
describe("discoverLocalTransport", () => { describe("discoverLocalTransport", () => {
@@ -53,9 +53,9 @@ describe("discoverLocalTransport", () => {
await expect( await expect(
discoverLocalTransport( discoverLocalTransport(
c, c,
backend(), "livekit",
[],
{ transport: hostTransport }, { transport: hostTransport },
fromBackends(),
logger, logger,
), ),
).resolves.toBe(hostTransport); ).resolves.toBe(hostTransport);
@@ -66,9 +66,9 @@ describe("discoverLocalTransport", () => {
await expect( await expect(
discoverLocalTransport( discoverLocalTransport(
client([homeserverTransport]), client([homeserverTransport]),
backend(), "livekit",
[],
{ fallbackTransport: hostTransport }, { fallbackTransport: hostTransport },
fromBackends(),
logger, logger,
), ),
).resolves.toEqual(homeserverTransport); ).resolves.toEqual(homeserverTransport);
@@ -78,44 +78,47 @@ describe("discoverLocalTransport", () => {
await expect( await expect(
discoverLocalTransport( discoverLocalTransport(
client([{ type: "not_livekit" }]), client([{ type: "not_livekit" }]),
backend(), "livekit",
[],
{ fallbackTransport: hostTransport }, { fallbackTransport: hostTransport },
fromBackends(),
logger, logger,
), ),
).resolves.toEqual(hostTransport); ).resolves.toEqual(hostTransport);
}); });
it("prefers a local transport over other members' SFUs", async () => { it("prefers a local transport over the backends' without asking them", async () => {
const b = backend(bobsTransport); const backends = fromBackends([bobsTransport]);
await expect( await expect(
discoverLocalTransport( discoverLocalTransport(
client([homeserverTransport]), client([homeserverTransport]),
b, "livekit",
[bob],
{}, {},
backends,
logger, logger,
), ),
).resolves.toEqual(homeserverTransport); ).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 () => { it("takes the first transport the backends provide where there is no local one", async () => {
const b = backend(bobsTransport);
await expect( await expect(
discoverLocalTransport(client([]), b, [bob], {}, logger), discoverLocalTransport(
client([]),
"livekit",
{},
fromBackends([bobsTransport, carolsTransport]),
logger,
),
).resolves.toEqual(bobsTransport); ).resolves.toEqual(bobsTransport);
expect(b.findRemotePublishTransport).toHaveBeenCalledWith([bob]);
}); });
it("fails naming the homeserver where there is no transport at all", async () => { it("fails naming the homeserver where there is no transport at all", async () => {
const promise = discoverLocalTransport( const promise = discoverLocalTransport(
client([]), client([]),
backend(), "livekit",
[bob],
{}, {},
fromBackends(),
logger, logger,
); );
+7 -10
View File
@@ -7,42 +7,39 @@ Please see LICENSE in the repository root for full details.
import { type MatrixClient } from "matrix-js-sdk"; import { type MatrixClient } from "matrix-js-sdk";
import { type Logger } from "matrix-js-sdk/lib/logger"; import { type Logger } from "matrix-js-sdk/lib/logger";
import { import { type Transport } from "matrix-js-sdk/lib/matrixrtc";
type CallMembership,
type Transport,
} from "matrix-js-sdk/lib/matrixrtc";
import { type RTCParticipationOptions } from "../api"; import { type RTCParticipationOptions } from "../api";
import { MatrixRTCTransportMissingError } from "../errors"; import { MatrixRTCTransportMissingError } from "../errors";
import { type MediaBackend } from "../media-backend/api";
import { TransportAutoDiscovery } from "./TransportAutoDiscovery"; import { TransportAutoDiscovery } from "./TransportAutoDiscovery";
/** /**
* The transport the local member advertises: the one the host names, else the * 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. * else the first the backends can provide, which are asked only then.
* *
* @throws {MatrixRTCTransportMissingError} where none of these exists. * @throws {MatrixRTCTransportMissingError} where none of these exists.
*/ */
export async function discoverLocalTransport( export async function discoverLocalTransport(
client: Pick<MatrixClient, "getDomain" | "_unstable_getRTCTransports">, client: Pick<MatrixClient, "getDomain" | "_unstable_getRTCTransports">,
backend: Pick<MediaBackend, "transportType" | "findRemotePublishTransport">, transportType: string,
memberships: CallMembership[],
{ {
transport: override, transport: override,
fallbackTransport, fallbackTransport,
}: Pick<RTCParticipationOptions, "transport" | "fallbackTransport">, }: Pick<RTCParticipationOptions, "transport" | "fallbackTransport">,
transportsFromBackends: () => Promise<Transport[]>,
logger: Logger, logger: Logger,
): Promise<Transport> { ): Promise<Transport> {
const transport = const transport =
override ?? override ??
(await new TransportAutoDiscovery({ (await new TransportAutoDiscovery({
client, client,
transportType: backend.transportType, transportType,
fallbackTransport, fallbackTransport,
logger, logger,
}).discoverPreferredTransport()) ?? }).discoverPreferredTransport()) ??
(await backend.findRemotePublishTransport(memberships)); (await transportsFromBackends()).at(0) ??
null;
if (transport === null) if (transport === null)
throw new MatrixRTCTransportMissingError(client.getDomain() ?? ""); throw new MatrixRTCTransportMissingError(client.getDomain() ?? "");
return transport; return transport;
+1 -15
View File
@@ -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 Logger } from "matrix-js-sdk/lib/logger";
import { import { type Transport } from "matrix-js-sdk/lib/matrixrtc";
type CallMembership,
type Transport,
} from "matrix-js-sdk/lib/matrixrtc";
import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager"; import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager";
import { type Observable } from "rxjs"; import { type Observable } from "rxjs";
@@ -119,17 +116,6 @@ export interface MediaBackend {
/** The membership transport type this backend serves: "livekit". */ /** The membership transport type this backend serves: "livekit". */
readonly transportType: string; 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<Transport | null>;
/** /**
* Makes `transport` usable as the one the local member publishes on. The * 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 * session found it; the backend does whatever has to succeed before it is
@@ -25,7 +25,6 @@ import { defaultSessionTimings } from "../../config";
import { import {
flushPromises, flushPromises,
mockEmitter, mockEmitter,
mockRtcMembership,
ownMemberMock, ownMemberMock,
testScope, testScope,
} from "../../utils/test"; } from "../../utils/test";
@@ -276,35 +275,6 @@ describe("createLivekitBackend", () => {
expect(connections[0].connection.serverName).toBe("sample.com"); 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 () => { it("waits for the publisher connection to hold its token before the join", async () => {
fetchMock.post(homeserverEndpoint, 404); fetchMock.post(homeserverEndpoint, 404);
fetchMock.post(`${exampleServiceUrl}/delegate_delayed_leave`, 404); fetchMock.post(`${exampleServiceUrl}/delegate_delayed_leave`, 404);
@@ -8,7 +8,6 @@ Please see LICENSE in the repository root for full details.
import { type MatrixClient } from "matrix-js-sdk"; import { type MatrixClient } from "matrix-js-sdk";
import { type Logger } from "matrix-js-sdk/lib/logger"; import { type Logger } from "matrix-js-sdk/lib/logger";
import { import {
type CallMembership,
isUnstableLivekitTransport, isUnstableLivekitTransport,
type Transport, type Transport,
} from "matrix-js-sdk/lib/matrixrtc"; } from "matrix-js-sdk/lib/matrixrtc";
@@ -63,7 +62,6 @@ import { createDataChannel$ } from "./DataChannel";
import { createKeyProvider } from "./KeyProvider"; import { createKeyProvider } from "./KeyProvider";
import { createLivekitMemberMedia } from "./LivekitMemberMedia"; import { createLivekitMemberMedia } from "./LivekitMemberMedia";
import { createLivekitLocalMedia } from "./LocalMedia"; import { createLivekitLocalMedia } from "./LocalMedia";
import { findRemotePublishTransport } from "./RemotePublishTransport";
export interface LivekitBackendOptions { export interface LivekitBackendOptions {
/** /**
@@ -150,18 +148,6 @@ export function createLivekitBackend(
const ownServerName = serverNameOf(ownMembershipIdentity.userId); const ownServerName = serverNameOf(ownMembershipIdentity.userId);
const remotePublishTransport = async (
memberships: CallMembership[],
): Promise<Transport | null> =>
findRemotePublishTransport({
client,
memberships,
ownMembershipIdentity,
roomId,
slotId,
logger,
});
const prepareLocalTransport = async ( const prepareLocalTransport = async (
transport: Transport, transport: Transport,
): Promise<TransportCapabilities> => { ): Promise<TransportCapabilities> => {
@@ -286,7 +272,6 @@ export function createLivekitBackend(
return { return {
transportType: "livekit", transportType: "livekit",
findRemotePublishTransport: remotePublishTransport,
prepareLocalTransport, prepareLocalTransport,
delegateDelayedLeave: delegateLeave, delegateDelayedLeave: delegateLeave,
local: localMedia_, local: localMedia_,
@@ -12,7 +12,7 @@ import { beforeEach, describe, expect, it, type Mocked, vi } from "vitest";
import { mockRtcMembership, ownMemberMock } from "../../utils/test"; import { mockRtcMembership, ownMemberMock } from "../../utils/test";
import { type ClientGetTokenParts } from "./auth/csApi"; 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. */ /** A JWT whose payload states whether the issuing server lets us publish. */
function jwtWithPublishGrant(canPublish: boolean): string { function jwtWithPublishGrant(canPublish: boolean): string {
@@ -27,6 +27,8 @@ function jwtWithPublishGrant(canPublish: boolean): string {
const bobSfu = { type: "livekit", url: "wss://sfu.bob.com" }; const bobSfu = { type: "livekit", url: "wss://sfu.bob.com" };
const carolSfu = { type: "livekit", url: "wss://sfu.carol.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 // Bob is the oldest member, Carol joined after him
const bob = mockRtcMembership("@bob:bob.com", "BOB", { const bob = mockRtcMembership("@bob:bob.com", "BOB", {
@@ -37,24 +39,41 @@ const carol = mockRtcMembership("@carol:carol.com", "CAROL", {
fociPreferred: [carolSfu], fociPreferred: [carolSfu],
membership: { created_ts: 200 }, 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: [ fociPreferred: [
{ type: "livekit", livekit_service_url: "https://jwt.dave.com" }, { type: "livekit", livekit_service_url: "https://jwt.erin.com" },
], ],
membership: { created_ts: 50 }, 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( const csApiMissing = new MatrixError(
{ errcode: "M_UNRECOGNIZED", error: "" }, { errcode: "M_UNRECOGNIZED", error: "" },
404, 404,
); );
describe("findRemotePublishTransport", () => { describe("findRemotePublishLivekitTransport", () => {
let client: Mocked<ClientGetTokenParts>; let client: Mocked<ClientGetTokenParts>;
const find = async ( const find = async (
memberships: CallMembership[], memberships: CallMembership[],
): ReturnType<typeof findRemotePublishTransport> => ): ReturnType<typeof findRemotePublishLivekitTransport> =>
findRemotePublishTransport({ findRemotePublishLivekitTransport({
client, client,
memberships, memberships,
ownMembershipIdentity: ownMemberMock, ownMembershipIdentity: ownMemberMock,
@@ -72,11 +91,7 @@ describe("findRemotePublishTransport", () => {
}); });
it("publishes on the oldest member's SFU with its server_name", async () => { it("publishes on the oldest member's SFU with its server_name", async () => {
await expect(find([carol, bob])).resolves.toEqual({ await expect(find([carol, bob])).resolves.toEqual(bobsTransport);
type: "livekit",
url: "wss://sfu.bob.com",
server_name: "bob.com",
});
expect(client._unstable_getLivekitToken).toHaveBeenCalledTimes(1); expect(client._unstable_getLivekitToken).toHaveBeenCalledTimes(1);
expect(client._unstable_getLivekitToken).toHaveBeenCalledWith({ expect(client._unstable_getLivekitToken).toHaveBeenCalledWith({
url: "wss://sfu.bob.com", url: "wss://sfu.bob.com",
@@ -87,10 +102,22 @@ describe("findRemotePublishTransport", () => {
}); });
}); });
it("ignores members publishing through a legacy JWT service", async () => { it("asks about an SFU once however many members are on it", async () => {
await expect(find([legacyDave, bob])).resolves.toMatchObject({ client._unstable_getLivekitToken.mockResolvedValue({
server_name: "bob.com", 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); expect(client._unstable_getLivekitToken).toHaveBeenCalledTimes(1);
}); });
@@ -102,9 +129,7 @@ describe("findRemotePublishTransport", () => {
}), }),
); );
await expect(find([bob, carol])).resolves.toMatchObject({ await expect(find([bob, carol])).resolves.toEqual(carolsTransport);
server_name: "carol.com",
});
expect(client._unstable_getLivekitToken).toHaveBeenCalledTimes(2); expect(client._unstable_getLivekitToken).toHaveBeenCalledTimes(2);
}); });
@@ -121,9 +146,7 @@ describe("findRemotePublishTransport", () => {
: Promise.resolve({ jwt: jwtWithPublishGrant(true) }), : Promise.resolve({ jwt: jwtWithPublishGrant(true) }),
); );
await expect(find([bob, carol])).resolves.toMatchObject({ await expect(find([bob, carol])).resolves.toEqual(carolsTransport);
server_name: "carol.com",
});
}); });
it("finds nothing once all candidates are exhausted", async () => { 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 () => { 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(); expect(client._unstable_getLivekitToken).not.toHaveBeenCalled();
}); });
}); });
@@ -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 Logger } from "matrix-js-sdk/lib/logger";
import { import {
type CallMembership, type CallMembership,
getRemoteLivekitPublishCandidates, isUnstableLivekitTransport,
type LivekitPublishCandidate,
type Transport, type Transport,
} from "matrix-js-sdk/lib/matrixrtc"; } from "matrix-js-sdk/lib/matrixrtc";
import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager"; 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 * 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 * transport to advertise, or null where none does. Candidates are tried
* are tried oldest membership first, and each one is checked by asking for a * oldest membership first, each SFU once, and each is checked by asking for a
* publisher token through our homeserver (MSC4195): a homeserver may restrict * publisher token through our homeserver (MSC4195): a homeserver may restrict
* remote publishers, which shows as a token without the publish grant. * remote publishers, which shows as a token without the publish grant.
*/ */
export async function findRemotePublishTransport({ export async function findRemotePublishLivekitTransport({
client, client,
memberships, memberships,
ownMembershipIdentity, ownMembershipIdentity: own,
roomId, roomId,
slotId, slotId,
logger, logger,
@@ -46,45 +45,51 @@ export async function findRemotePublishTransport({
logger.info( logger.info(
"No local transport available, looking for another member's SFU to publish on", "No local transport available, looking for another member's SFU to publish on",
); );
const candidates = getRemoteLivekitPublishCandidates( const tried = new Set<string>();
memberships, const byAge = [...memberships].sort((a, b) => a.createdTs() - b.createdTs());
ownMembershipIdentity, for (const membership of byAge) {
); if (
for (const candidate of candidates) { membership.userId === own.userId &&
const outcome = await tryCandidate(candidate); membership.deviceId === own.deviceId
if (outcome === "usable") return candidate.transport; )
if (outcome === "noCsApi") return null; 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; 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";
}
} }