Fall back to publishing on a remote SFU if possible

This commit is contained in:
Johannes Marbach
2026-10-14 09:45:44 +02:00
parent b158c4fcd7
commit edf1feae69
15 changed files with 446 additions and 93 deletions
+1 -1
View File
@@ -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.", "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": "Failed to connect to Livekit server",
"livekit_connection_error_description": "An error occurred while connecting to the Livekit server (<1>Reason:</1> <2>{{ reason }}</2>).", "livekit_connection_error_description": "An error occurred while connecting to the Livekit server (<1>Reason:</1> <2>{{ reason }}</2>).",
"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": "Membership Manager Error",
"membership_manager_description": "The Membership Manager had to shut down. This is caused by many consecutive failed network requests.", "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.", "no_matrix_2_authorization_service": "The authorization service for your media server (SFU) is out of date.",
+1 -1
View File
@@ -103,7 +103,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#develop", "matrix-js-sdk": "github:matrix-org/matrix-js-sdk#9405287c22797e9d69acd366c47bf412fa4384c8",
"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",
+6 -6
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#develop specifier: github:matrix-org/matrix-js-sdk#9405287c22797e9d69acd366c47bf412fa4384c8
version: https://codeload.github.com/matrix-org/matrix-js-sdk/tar.gz/4df2f2bfb5826c25f05d9535a8556acecaf2e570 version: https://codeload.github.com/matrix-org/matrix-js-sdk/tar.gz/9405287c22797e9d69acd366c47bf412fa4384c8
matrix-widget-api: matrix-widget-api:
specifier: ^1.18.0 specifier: ^1.18.0
version: 1.20.0 version: 1.20.0
@@ -4673,9 +4673,9 @@ 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/4df2f2bfb5826c25f05d9535a8556acecaf2e570: matrix-js-sdk@https://codeload.github.com/matrix-org/matrix-js-sdk/tar.gz/9405287c22797e9d69acd366c47bf412fa4384c8:
resolution: {gitHosted: true, integrity: sha512-b1o6ywaPY9jakj+3GIgx3samM9AbqohzPQ5nZKbyOGYB16IHQtNMyjEo/PY0fW/s0uwBVApEHNb/MMtN40c/rQ==, tarball: https://codeload.github.com/matrix-org/matrix-js-sdk/tar.gz/4df2f2bfb5826c25f05d9535a8556acecaf2e570} resolution: {gitHosted: true, integrity: sha512-pYk38ggTEdM0WWXLDiFoKmyRn3wo3o8m1ge0iCtxjBmgAW5qoQ2Ifd8oHVy+gu6PQe/CfLuKPXEXH8hff9C8GA==, tarball: https://codeload.github.com/matrix-org/matrix-js-sdk/tar.gz/9405287c22797e9d69acd366c47bf412fa4384c8}
version: 43.0.0 version: 44.0.0
engines: {node: '>=22.0.0'} engines: {node: '>=22.0.0'}
matrix-widget-api@1.20.0: matrix-widget-api@1.20.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/4df2f2bfb5826c25f05d9535a8556acecaf2e570: matrix-js-sdk@https://codeload.github.com/matrix-org/matrix-js-sdk/tar.gz/9405287c22797e9d69acd366c47bf412fa4384c8:
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
+1 -1
View File
@@ -45,7 +45,7 @@ function isEndpointUnsupported(e: unknown): boolean {
* Matrix error. A connection error is neither, as the endpoint may well exist. * Matrix error. A connection error is neither, as the endpoint may well exist.
* @param e The error to check. * @param e The error to check.
*/ */
function isCsApiUnavailable(e: unknown): boolean { export function isCsApiUnavailable(e: unknown): boolean {
if (isEndpointUnsupported(e)) return true; if (isEndpointUnsupported(e)) return true;
return !(e instanceof MatrixError) && !(e instanceof ConnectionError); return !(e instanceof MatrixError) && !(e instanceof ConnectionError);
} }
+10 -3
View File
@@ -36,7 +36,7 @@ export interface SFUConfig {
/** /**
* Decoded details from the JWT. * Decoded details from the JWT.
*/ */
interface SFUJWTPayload { export interface SFUJWTPayload {
/** /**
* Expiration time for the JWT. * Expiration time for the JWT.
* Note: This value is in seconds since Unix epoch. * Note: This value is in seconds since Unix epoch.
@@ -76,8 +76,7 @@ export function extractFullConfigFromToken(sfuConfig: {
url: string; url: string;
jwt: string; jwt: string;
}): SFUConfig { }): SFUConfig {
const [, payloadStr] = sfuConfig.jwt.split("."); const payload = decodeSFUJWTPayload(sfuConfig.jwt);
const payload = JSON.parse(global.atob(payloadStr)) as SFUJWTPayload;
return { return {
jwt: sfuConfig.jwt, jwt: sfuConfig.jwt,
url: sfuConfig.url, url: sfuConfig.url,
@@ -88,3 +87,11 @@ export function extractFullConfigFromToken(sfuConfig: {
livekitIdentity: payload.sub, 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;
}
+1 -1
View File
@@ -527,7 +527,7 @@ export const DeveloperSettingsTab: FC<Props> = ({
client, client,
membership: { userId, deviceId, memberId: "" }, membership: { userId, deviceId, memberId: "" },
transport, transport,
serverName: client.getDomain()!, serverName: transport.server_name ?? client.getDomain()!,
roomId, roomId,
slotId: "org.example.fake-slot#room", slotId: "org.example.fake-slot#room",
role: "publisher", role: "publisher",
+12 -1
View File
@@ -567,7 +567,18 @@ export function createCallViewModel$(
const localTransport$ = options.localTransport const localTransport$ = options.localTransport
? constant(options.localTransport) ? constant(options.localTransport)
: from(getLocalTransport(client)); : from(
getLocalTransport({
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: matrixRTCSession.memberships,
ownMembershipIdentity,
roomId: matrixRoom.roomId,
slotId: matrixRTCSession.slotId,
logger,
}),
);
const connectionFactory = const connectionFactory =
options.connectionFactory ?? options.connectionFactory ??
@@ -427,6 +427,56 @@ describe("LocalMembership", () => {
}, },
); );
it("joins without delegation when publishing on a remote SFU", async () => {
// Our homeserver only delegates to its own SFUs (MSC4195), so publishing
// on another homeserver's SFU means restarting the delayed leave ourselves.
const scope = testScope();
const joinMatrixRTC = vi.fn();
const delayId$ = new BehaviorSubject<string | null>(null);
const remoteTransport: UnstableLivekitTransport = {
type: "livekit",
url: "https://sfu.sample.com",
server_name: "sample.com",
};
const connectionMap = new ConnectionMap();
connectionMap.add(
{
...connectionTransportAConnected,
transport: remoteTransport,
serverName: "sample.com",
} as Connection,
[],
);
// Our own homeserver does support delegation, so only the SFU being
// remote can explain delegation being skipped
fetchMock.post(
"https://matrix.example.org/_matrix/client/unstable/io.element.msc4195/rtc/livekit/delegate_delayed_leave",
() => ({ status: 401, body: {} }),
);
const localMembership = createLocalMembership$({
scope,
...defaultCreateLocalMemberValues,
connectionMap$: constant(new Epoch(connectionMap)),
joinMatrixRTC,
localTransport$: constant({
transport: remoteTransport,
serverName: "sample.com",
}),
delayId$,
});
localMembership.requestJoinAndPublish();
void (await fetchMock.flush());
await flushPromises();
expect(joinMatrixRTC).toHaveBeenCalledWith(remoteTransport, timings);
delayId$.next("leave1");
await flushPromises();
expect(delegationSpy).not.toHaveBeenCalled();
});
it("only starts tracks if requested", async () => { it("only starts tracks if requested", async () => {
const scope = new ObservableScope(); const scope = new ObservableScope();
@@ -43,6 +43,7 @@ import {
Subject, Subject,
} from "rxjs"; } from "rxjs";
import { type Logger } from "matrix-js-sdk/lib/logger"; import { type Logger } from "matrix-js-sdk/lib/logger";
import { type MatrixClient } from "matrix-js-sdk";
import { deepCompare } from "matrix-js-sdk/lib/utils"; import { deepCompare } from "matrix-js-sdk/lib/utils";
import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager"; import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager";
@@ -154,7 +155,9 @@ interface Props {
roomId: string; roomId: string;
ownMembershipIdentity: CallMembershipIdentityParts; ownMembershipIdentity: CallMembershipIdentityParts;
localTransport$: Observable<TransportLocator>; localTransport$: Observable<TransportLocator>;
client: ClientDelegationParts & ClientOpenIDParts; client: ClientDelegationParts &
ClientOpenIDParts &
Pick<MatrixClient, "getDomain">;
matrixRTCSession: Pick< matrixRTCSession: Pick<
MatrixRTCSession, MatrixRTCSession,
"slotId" | "updateCallIntent" | "leave" "slotId" | "updateCallIntent" | "leave"
@@ -312,17 +315,26 @@ export const createLocalMembership$ = ({
// to whether delayed event delegation is supported // to whether delayed event delegation is supported
const joinParams$ = scope.behavior( const joinParams$ = scope.behavior(
localTransport$.pipe( localTransport$.pipe(
switchMap(async ({ transport }) => { switchMap(async ({ transport, serverName }) => {
// Delegation goes through whichever service authenticates us as a let delegationSupported: boolean;
// publisher, so only that service's support counts if (serverName !== client.getDomain()) {
const flow = chooseAuthFlow(transport, "publisher"); // Delegation is only possible when publishing on a local SFU.
const delegationSupported = logger.info(
flow.kind === "openID" `Publishing on a remote SFU of ${serverName}, delegation is not possible`,
? await checkDelegationSupport( );
flow.serviceUrl + "/delegate_delayed_leave", delegationSupported = false;
`transport ${flow.serviceUrl}`, } else {
) // Delegation goes through whichever service authenticates us as a
: await homeserverSupportsDelegation; // publisher, so only that service's support counts
const flow = chooseAuthFlow(transport, "publisher");
delegationSupported =
flow.kind === "openID"
? await checkDelegationSupport(
flow.serviceUrl + "/delegate_delayed_leave",
`transport ${flow.serviceUrl}`,
)
: await homeserverSupportsDelegation;
}
return { transport, delegationSupported }; return { transport, delegationSupported };
}), }),
), ),
@@ -15,68 +15,94 @@ import {
vi, vi,
} from "vitest"; } from "vitest";
import fetchMock from "fetch-mock"; import fetchMock from "fetch-mock";
import { MatrixError } from "matrix-js-sdk";
import { type CallMembership } from "matrix-js-sdk/lib/matrixrtc";
import { mockConfig } from "../../../utils/test"; import {
import { getLocalTransport } from "./LocalTransport"; mockConfig,
mockRtcMembership,
ownMemberMock,
} from "../../../utils/test";
import { getLocalTransport, type LocalTransportClient } from "./LocalTransport";
import { MatrixRTCTransportMissingError } from "../../../utils/errors"; import { MatrixRTCTransportMissingError } from "../../../utils/errors";
import { customTransport } from "../../../settings/settings"; import { customTransport } from "../../../settings/settings";
import { type MatrixClient } from "matrix-js-sdk";
/**
* Builds 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 },
});
describe("LocalTransport", () => { describe("LocalTransport", () => {
beforeEach(() => vi.clearAllMocks()); let client: MockedObject<LocalTransportClient>;
const params = (
memberships: CallMembership[] = [],
): Parameters<typeof getLocalTransport>[0] => ({
client,
memberships,
ownMembershipIdentity: ownMemberMock,
roomId: "!room:example.org",
slotId: "m.call#ROOM",
});
beforeEach(() => {
vi.clearAllMocks();
mockConfig({});
customTransport.setValue(customTransport.defaultValue);
client = {
getDomain: vi.fn().mockReturnValue("example.org"),
_unstable_getRTCTransports: vi.fn().mockResolvedValue([]),
_unstable_getLivekitToken: vi
.fn()
.mockRejectedValue(
new MatrixError({ errcode: "M_UNRECOGNIZED", error: "" }, 404),
),
};
});
afterEach(() => {
fetchMock.reset();
});
it("throws if config is missing", async () => { it("throws if config is missing", async () => {
await expect( await expect(getLocalTransport(params())).rejects.toThrow(
getLocalTransport({ new MatrixRTCTransportMissingError("example.org"),
_unstable_getRTCTransports: async () => Promise.resolve([]), );
getDomain: () => "example.org",
}),
).rejects.toThrow(new MatrixRTCTransportMissingError("example.org"));
}); });
it("returns preferred transport", async () => { describe("local transport configuration mechanisms", () => {
// Use config so transport discovery succeeds
mockConfig({
livekit: { livekit_service_url: "https://lk.example.org" },
});
expect(
await getLocalTransport({
_unstable_getRTCTransports: async () => Promise.resolve([]),
getDomain: () => "example.org",
}),
).toStrictEqual({
transport: {
livekit_service_url: "https://lk.example.org",
type: "livekit",
},
serverName: "example.org",
});
});
describe("transport configuration mechanisms", () => {
let client: MockedObject<
Pick<MatrixClient, "getDomain" | "_unstable_getRTCTransports">
>;
beforeEach(() => {
mockConfig({});
customTransport.setValue(customTransport.defaultValue);
client = {
getDomain: vi.fn().mockReturnValue("example.org"),
_unstable_getRTCTransports: vi.fn().mockResolvedValue([]),
};
});
afterEach(() => {
fetchMock.reset();
});
it("supports getting transport via application config", async () => { it("supports getting transport via application config", async () => {
mockConfig({ mockConfig({
livekit: { livekit_service_url: "https://lk.example.org" }, livekit: { livekit_service_url: "https://lk.example.org" },
}); });
expect(await getLocalTransport(client)).toStrictEqual({ expect(await getLocalTransport(params())).toStrictEqual({
transport: { transport: {
livekit_service_url: "https://lk.example.org", livekit_service_url: "https://lk.example.org",
type: "livekit", type: "livekit",
@@ -91,7 +117,7 @@ describe("LocalTransport", () => {
livekit_service_url: "https://lk.example.org", livekit_service_url: "https://lk.example.org",
}); });
expect(await getLocalTransport(client)).toStrictEqual({ expect(await getLocalTransport(params())).toStrictEqual({
transport: { transport: {
livekit_service_url: "https://lk.example.org", livekit_service_url: "https://lk.example.org",
type: "livekit", type: "livekit",
@@ -100,12 +126,29 @@ describe("LocalTransport", () => {
}); });
}); });
it("locates a custom transport with server_name on that homeserver", async () => {
customTransport.setValue({
type: "livekit",
url: "wss://sfu.bob.com",
server_name: "bob.com",
});
expect(await getLocalTransport(params())).toStrictEqual({
transport: {
type: "livekit",
url: "wss://sfu.bob.com",
server_name: "bob.com",
},
serverName: "bob.com",
});
});
it("supports getting transport via backend", async () => { it("supports getting transport via backend", async () => {
client._unstable_getRTCTransports.mockResolvedValue([ client._unstable_getRTCTransports.mockResolvedValue([
{ type: "livekit", livekit_service_url: "https://lk.example.org" }, { type: "livekit", livekit_service_url: "https://lk.example.org" },
]); ]);
expect(await getLocalTransport(client)).toStrictEqual({ expect(await getLocalTransport(params())).toStrictEqual({
transport: { transport: {
livekit_service_url: "https://lk.example.org", livekit_service_url: "https://lk.example.org",
type: "livekit", type: "livekit",
@@ -114,13 +157,110 @@ describe("LocalTransport", () => {
}); });
}); });
it("prefers a local transport over other members' SFUs", async () => {
client._unstable_getRTCTransports.mockResolvedValue([
{ type: "livekit", url: "wss://sfu.example.org" },
]);
expect(await getLocalTransport(params([bob]))).toStrictEqual({
transport: { type: "livekit", url: "wss://sfu.example.org" },
serverName: "example.org",
});
expect(client._unstable_getLivekitToken).not.toHaveBeenCalled();
});
it("throws if no options are available", async () => { it("throws if no options are available", async () => {
await expect( await expect(getLocalTransport(params())).rejects.toThrow(
getLocalTransport({ new MatrixRTCTransportMissingError("example.org"),
getDomain: () => "example.org", );
_unstable_getRTCTransports: async () => Promise.resolve([]), });
}), });
).rejects.toThrow(new MatrixRTCTransportMissingError("example.org"));
describe("publishing on another member's SFU", () => {
beforeEach(() => {
client._unstable_getLivekitToken.mockResolvedValue({
jwt: jwtWithPublishGrant(true),
});
});
it("publishes on the oldest member's SFU with its server_name", async () => {
expect(await getLocalTransport(params([carol, bob]))).toStrictEqual({
transport: {
type: "livekit",
url: "wss://sfu.bob.com",
server_name: "bob.com",
},
serverName: "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 () => {
expect(await getLocalTransport(params([legacyDave, bob]))).toMatchObject({
serverName: "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"),
}),
);
expect(await getLocalTransport(params([bob, carol]))).toMatchObject({
serverName: "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) }),
);
expect(await getLocalTransport(params([bob, carol]))).toMatchObject({
serverName: "carol.com",
});
});
it("throws once all candidates are exhausted", async () => {
client._unstable_getLivekitToken.mockResolvedValue({
jwt: jwtWithPublishGrant(false),
});
await expect(getLocalTransport(params([bob, carol]))).rejects.toThrow(
new MatrixRTCTransportMissingError("example.org"),
);
expect(client._unstable_getLivekitToken).toHaveBeenCalledTimes(2);
});
it("gives up immediately if our homeserver lacks the CS API", async () => {
client._unstable_getLivekitToken.mockRejectedValue(
new MatrixError({ errcode: "M_UNRECOGNIZED", error: "" }, 404),
);
await expect(getLocalTransport(params([bob, carol]))).rejects.toThrow(
new MatrixRTCTransportMissingError("example.org"),
);
expect(client._unstable_getLivekitToken).toHaveBeenCalledTimes(1);
}); });
}); });
}); });
@@ -6,28 +6,78 @@ 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 { logger as rootLogger } from "matrix-js-sdk/lib/logger"; import { logger as rootLogger, type Logger } from "matrix-js-sdk/lib/logger";
import {
type CallMembership,
getRemoteLivekitPublishCandidates,
type LivekitPublishCandidate,
} from "matrix-js-sdk/lib/matrixrtc";
import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager";
import { Config } from "../../../config/Config.ts"; import { Config } from "../../../config/Config.ts";
import { MatrixRTCTransportMissingError } from "../../../utils/errors.ts"; import { MatrixRTCTransportMissingError } from "../../../utils/errors.ts";
import { customTransport as customTransportSetting } from "../../../settings/settings.ts"; import { customTransport as customTransportSetting } from "../../../settings/settings.ts";
import { RtcTransportAutoDiscovery } from "./RtcTransportAutoDiscovery.ts"; import { RtcTransportAutoDiscovery } from "./RtcTransportAutoDiscovery.ts";
import { type TransportLocator } from "../../../livekit/auth/types.ts"; import {
type TransportLocator,
isCsApiUnavailable,
} from "../../../livekit/auth";
import * as csApi from "../../../livekit/auth/csApi.ts";
import { decodeSFUJWTPayload } from "../../../livekit/auth/types.ts";
export type LocalTransportClient = Pick<
MatrixClient,
"getDomain" | "_unstable_getRTCTransports" | "_unstable_getLivekitToken"
>;
export interface LocalTransportParams {
client: LocalTransportClient;
/**
* The current members of the session. If our own homeserver provides no
* transport, we fall back to publishing on one of theirs.
*/
memberships: CallMembership[];
/**
* Data identifying our own session membership, needed to request tokens.
*/
ownMembershipIdentity: CallMembershipIdentityParts;
/**
* The ID of the Matrix room in which the session takes place.
*/
roomId: string;
/**
* The ID of the MatrixRTC slot in which the session takes place.
*/
slotId: string;
logger?: Logger;
}
/** /**
* Determines the transport to advertise in our MatrixRTC membership and publish * Determines the transport to advertise in our MatrixRTC membership and publish
* media on. * media on.
*
* This tries to fall back to the SFU of another session member if needed.
*
* @throws {MatrixRTCTransportMissingError} if neither a local nor a remote
* transport is available.
*/ */
export async function getLocalTransport( export async function getLocalTransport({
client: Pick<MatrixClient, "getDomain" | "_unstable_getRTCTransports">, client,
): Promise<TransportLocator> { memberships,
const serverName = client.getDomain(); ownMembershipIdentity,
if (serverName === null) throw new Error("No server name"); roomId,
slotId,
logger: parentLogger = rootLogger,
}: LocalTransportParams): Promise<TransportLocator> {
const logger = parentLogger.getChild("[LocalTransport]");
const domain = client.getDomain();
if (domain === null) throw new Error("No server name");
const ownServerName: string = domain;
const discovery = new RtcTransportAutoDiscovery({ const discovery = new RtcTransportAutoDiscovery({
client, client,
resolvedConfig: Config.get(), resolvedConfig: Config.get(),
logger: rootLogger.getChild("[LocalTransport]"), logger,
}); });
// Respect the user's custom transport, if set // Respect the user's custom transport, if set
@@ -35,7 +85,64 @@ export async function getLocalTransport(
customTransportSetting.value$.value ?? customTransportSetting.value$.value ??
(await discovery.discoverPreferredTransport()); (await discovery.discoverPreferredTransport());
if (transport === null) throw new MatrixRTCTransportMissingError(serverName); if (transport !== null) {
// A custom transport may point at another homeserver's SFU
const serverName =
"server_name" in transport && transport.server_name !== undefined
? transport.server_name
: ownServerName;
return { transport, serverName };
}
return { transport, serverName }; 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 locator = await tryRemoteCandidate(candidate);
if (locator !== null) return locator;
}
throw new MatrixRTCTransportMissingError(ownServerName);
/**
* Checks whether the homeserver behind a candidate SFU lets us publish on
* it. Servers MAY restrict remote publishers, in which case the token comes
* back without the publish grant and we move on to the next candidate.
* @returns The locator to publish with, or null if this candidate is unusable.
* @throws {MatrixRTCTransportMissingError} if the client-server API is not
* available at all, as no candidate can work then.
*/
async function tryRemoteCandidate({
url,
serverName,
transport,
}: LivekitPublishCandidate): Promise<TransportLocator | null> {
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 { transport, serverName };
}
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`,
);
throw new MatrixRTCTransportMissingError(ownServerName);
}
logger.warn(`Failed to get a token for ${url} of ${serverName}:`, e);
}
return null;
}
} }
+20
View File
@@ -83,6 +83,26 @@ describe("SessionBehaviors", () => {
).toEqual(["example.org", "sample.com"]); ).toEqual(["example.org", "sample.com"]);
}); });
it("locates a remote publisher's transport on the SFU's homeserver", () => {
// Dave has no SFU of his own and publishes on sample.com's
const remoteTransport: UnstableLivekitTransport = {
...urlTransport,
server_name: "sample.com",
};
const dave = mockRtcMembership("@dave:example.org", "DEV222", {
fociPreferred: [remoteTransport],
});
const { transports$ } = membershipsAndTransports$(
testScope(),
constant(new Epoch([dave])),
);
expect(transports$.value.value).toEqual([
{ transport: remoteTransport, serverName: "sample.com" },
]);
});
it("leaves out members without a LiveKit transport", () => { it("leaves out members without a LiveKit transport", () => {
const dave = mockRtcMembership("@dave:example.org", "DEV222", { const dave = mockRtcMembership("@dave:example.org", "DEV222", {
fociPreferred: [{ type: "something-else" } as Transport], fociPreferred: [{ type: "something-else" } as Transport],
+3 -1
View File
@@ -51,7 +51,9 @@ export const membershipsAndTransports$ = (
transport: isUnstableLivekitTransport(transport) transport: isUnstableLivekitTransport(transport)
? { ? {
transport, transport,
serverName: membership.userId.replace(/^.*?:/, ""), // The SFU's own homeserver, which is not necessarily the
// member's if they publish on a remote SFU (MSC4195)
serverName: membership.getTransportServerName(transport),
} }
: undefined, : undefined,
}; };
+2
View File
@@ -111,6 +111,8 @@ describe("widget", () => {
receiveToDevice: sendRecvToDevice, receiveToDevice: sendRecvToDevice,
turnServers: false, turnServers: false,
rtcTransports: true, rtcTransports: true,
rtcLivekitGetToken: true,
rtcLivekitDelegateDelayedLeave: true,
sendDelayedEvents: true, sendDelayedEvents: true,
updateDelayedEvents: true, updateDelayedEvents: true,
sendSticky: true, sendSticky: true,
+2
View File
@@ -175,6 +175,8 @@ export const initializeWidget = (
receiveToDevice: sendRecvToDevice, receiveToDevice: sendRecvToDevice,
turnServers: false, turnServers: false,
rtcTransports: true, rtcTransports: true,
rtcLivekitGetToken: true,
rtcLivekitDelegateDelayedLeave: true,
sendDelayedEvents: true, sendDelayedEvents: true,
updateDelayedEvents: true, updateDelayedEvents: true,
sendSticky: true, sendSticky: true,