mirror of
https://github.com/vector-im/element-call.git
synced 2026-10-16 23:45:50 +00:00
Fall back to publishing on a remote SFU if possible
This commit is contained in:
+1
-1
@@ -106,7 +106,7 @@
|
||||
"insufficient_capacity_description": "The server has reached its maximum capacity and you cannot join the call at this time. Try again later, or contact your server admin if the problem persists.",
|
||||
"livekit_connection_error": "Failed to connect to Livekit server",
|
||||
"livekit_connection_error_description": "An error occurred while connecting to the Livekit server (<1>Reason:</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_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.",
|
||||
|
||||
+1
-1
@@ -103,7 +103,7 @@
|
||||
"livekit-client": "^2.18.1",
|
||||
"lodash-es": "^4.17.21",
|
||||
"loglevel": "^1.9.1",
|
||||
"matrix-js-sdk": "github:matrix-org/matrix-js-sdk#develop",
|
||||
"matrix-js-sdk": "github:matrix-org/matrix-js-sdk#9405287c22797e9d69acd366c47bf412fa4384c8",
|
||||
"matrix-widget-api": "^1.18.0",
|
||||
"node-stdlib-browser": "^1.3.1",
|
||||
"normalize.css": "^8.0.1",
|
||||
|
||||
Generated
+6
-6
@@ -191,8 +191,8 @@ importers:
|
||||
specifier: ^1.9.1
|
||||
version: 1.9.2
|
||||
matrix-js-sdk:
|
||||
specifier: github:matrix-org/matrix-js-sdk#develop
|
||||
version: https://codeload.github.com/matrix-org/matrix-js-sdk/tar.gz/4df2f2bfb5826c25f05d9535a8556acecaf2e570
|
||||
specifier: github:matrix-org/matrix-js-sdk#9405287c22797e9d69acd366c47bf412fa4384c8
|
||||
version: https://codeload.github.com/matrix-org/matrix-js-sdk/tar.gz/9405287c22797e9d69acd366c47bf412fa4384c8
|
||||
matrix-widget-api:
|
||||
specifier: ^1.18.0
|
||||
version: 1.20.0
|
||||
@@ -4673,9 +4673,9 @@ packages:
|
||||
matrix-events-sdk@0.0.1:
|
||||
resolution: {integrity: sha512-1QEOsXO+bhyCroIe2/A5OwaxHvBm7EsSQ46DEDn8RBIfQwN5HWBpFvyWWR4QY0KHPPnnJdI99wgRiAl7Ad5qaA==}
|
||||
|
||||
matrix-js-sdk@https://codeload.github.com/matrix-org/matrix-js-sdk/tar.gz/4df2f2bfb5826c25f05d9535a8556acecaf2e570:
|
||||
resolution: {gitHosted: true, integrity: sha512-b1o6ywaPY9jakj+3GIgx3samM9AbqohzPQ5nZKbyOGYB16IHQtNMyjEo/PY0fW/s0uwBVApEHNb/MMtN40c/rQ==, tarball: https://codeload.github.com/matrix-org/matrix-js-sdk/tar.gz/4df2f2bfb5826c25f05d9535a8556acecaf2e570}
|
||||
version: 43.0.0
|
||||
matrix-js-sdk@https://codeload.github.com/matrix-org/matrix-js-sdk/tar.gz/9405287c22797e9d69acd366c47bf412fa4384c8:
|
||||
resolution: {gitHosted: true, integrity: sha512-pYk38ggTEdM0WWXLDiFoKmyRn3wo3o8m1ge0iCtxjBmgAW5qoQ2Ifd8oHVy+gu6PQe/CfLuKPXEXH8hff9C8GA==, tarball: https://codeload.github.com/matrix-org/matrix-js-sdk/tar.gz/9405287c22797e9d69acd366c47bf412fa4384c8}
|
||||
version: 44.0.0
|
||||
engines: {node: '>=22.0.0'}
|
||||
|
||||
matrix-widget-api@1.20.0:
|
||||
@@ -10227,7 +10227,7 @@ snapshots:
|
||||
|
||||
matrix-events-sdk@0.0.1: {}
|
||||
|
||||
matrix-js-sdk@https://codeload.github.com/matrix-org/matrix-js-sdk/tar.gz/4df2f2bfb5826c25f05d9535a8556acecaf2e570:
|
||||
matrix-js-sdk@https://codeload.github.com/matrix-org/matrix-js-sdk/tar.gz/9405287c22797e9d69acd366c47bf412fa4384c8:
|
||||
dependencies:
|
||||
'@babel/runtime': 8.0.0
|
||||
'@matrix-org/matrix-sdk-crypto-wasm': 18.9.0
|
||||
|
||||
@@ -45,7 +45,7 @@ function isEndpointUnsupported(e: unknown): boolean {
|
||||
* Matrix error. A connection error is neither, as the endpoint may well exist.
|
||||
* @param e The error to check.
|
||||
*/
|
||||
function isCsApiUnavailable(e: unknown): boolean {
|
||||
export function isCsApiUnavailable(e: unknown): boolean {
|
||||
if (isEndpointUnsupported(e)) return true;
|
||||
return !(e instanceof MatrixError) && !(e instanceof ConnectionError);
|
||||
}
|
||||
|
||||
@@ -36,7 +36,7 @@ export interface SFUConfig {
|
||||
/**
|
||||
* Decoded details from the JWT.
|
||||
*/
|
||||
interface SFUJWTPayload {
|
||||
export interface SFUJWTPayload {
|
||||
/**
|
||||
* Expiration time for the JWT.
|
||||
* Note: This value is in seconds since Unix epoch.
|
||||
@@ -76,8 +76,7 @@ export function extractFullConfigFromToken(sfuConfig: {
|
||||
url: string;
|
||||
jwt: string;
|
||||
}): SFUConfig {
|
||||
const [, payloadStr] = sfuConfig.jwt.split(".");
|
||||
const payload = JSON.parse(global.atob(payloadStr)) as SFUJWTPayload;
|
||||
const payload = decodeSFUJWTPayload(sfuConfig.jwt);
|
||||
return {
|
||||
jwt: sfuConfig.jwt,
|
||||
url: sfuConfig.url,
|
||||
@@ -88,3 +87,11 @@ export function extractFullConfigFromToken(sfuConfig: {
|
||||
livekitIdentity: payload.sub,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Decodes the payload of an SFU JWT, without verifying its signature.
|
||||
*/
|
||||
export function decodeSFUJWTPayload(jwt: string): SFUJWTPayload {
|
||||
const [, payloadStr] = jwt.split(".");
|
||||
return JSON.parse(global.atob(payloadStr)) as SFUJWTPayload;
|
||||
}
|
||||
|
||||
@@ -527,7 +527,7 @@ export const DeveloperSettingsTab: FC<Props> = ({
|
||||
client,
|
||||
membership: { userId, deviceId, memberId: "" },
|
||||
transport,
|
||||
serverName: client.getDomain()!,
|
||||
serverName: transport.server_name ?? client.getDomain()!,
|
||||
roomId,
|
||||
slotId: "org.example.fake-slot#room",
|
||||
role: "publisher",
|
||||
|
||||
@@ -567,7 +567,18 @@ export function createCallViewModel$(
|
||||
|
||||
const localTransport$ = 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 =
|
||||
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 () => {
|
||||
const scope = new ObservableScope();
|
||||
|
||||
|
||||
@@ -43,6 +43,7 @@ import {
|
||||
Subject,
|
||||
} from "rxjs";
|
||||
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 { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager";
|
||||
|
||||
@@ -154,7 +155,9 @@ interface Props {
|
||||
roomId: string;
|
||||
ownMembershipIdentity: CallMembershipIdentityParts;
|
||||
localTransport$: Observable<TransportLocator>;
|
||||
client: ClientDelegationParts & ClientOpenIDParts;
|
||||
client: ClientDelegationParts &
|
||||
ClientOpenIDParts &
|
||||
Pick<MatrixClient, "getDomain">;
|
||||
matrixRTCSession: Pick<
|
||||
MatrixRTCSession,
|
||||
"slotId" | "updateCallIntent" | "leave"
|
||||
@@ -312,17 +315,26 @@ export const createLocalMembership$ = ({
|
||||
// to whether delayed event delegation is supported
|
||||
const joinParams$ = scope.behavior(
|
||||
localTransport$.pipe(
|
||||
switchMap(async ({ transport }) => {
|
||||
switchMap(async ({ transport, serverName }) => {
|
||||
let delegationSupported: boolean;
|
||||
if (serverName !== client.getDomain()) {
|
||||
// Delegation is only possible when publishing on a local SFU.
|
||||
logger.info(
|
||||
`Publishing on a remote SFU of ${serverName}, delegation is not possible`,
|
||||
);
|
||||
delegationSupported = false;
|
||||
} else {
|
||||
// Delegation goes through whichever service authenticates us as a
|
||||
// publisher, so only that service's support counts
|
||||
const flow = chooseAuthFlow(transport, "publisher");
|
||||
const delegationSupported =
|
||||
delegationSupported =
|
||||
flow.kind === "openID"
|
||||
? await checkDelegationSupport(
|
||||
flow.serviceUrl + "/delegate_delayed_leave",
|
||||
`transport ${flow.serviceUrl}`,
|
||||
)
|
||||
: await homeserverSupportsDelegation;
|
||||
}
|
||||
return { transport, delegationSupported };
|
||||
}),
|
||||
),
|
||||
|
||||
@@ -15,55 +15,74 @@ import {
|
||||
vi,
|
||||
} from "vitest";
|
||||
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 { getLocalTransport } from "./LocalTransport";
|
||||
import {
|
||||
mockConfig,
|
||||
mockRtcMembership,
|
||||
ownMemberMock,
|
||||
} from "../../../utils/test";
|
||||
import { getLocalTransport, type LocalTransportClient } from "./LocalTransport";
|
||||
import { MatrixRTCTransportMissingError } from "../../../utils/errors";
|
||||
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", () => {
|
||||
beforeEach(() => vi.clearAllMocks());
|
||||
|
||||
it("throws if config is missing", async () => {
|
||||
await expect(
|
||||
getLocalTransport({
|
||||
_unstable_getRTCTransports: async () => Promise.resolve([]),
|
||||
getDomain: () => "example.org",
|
||||
}),
|
||||
).rejects.toThrow(new MatrixRTCTransportMissingError("example.org"));
|
||||
let client: MockedObject<LocalTransportClient>;
|
||||
const params = (
|
||||
memberships: CallMembership[] = [],
|
||||
): Parameters<typeof getLocalTransport>[0] => ({
|
||||
client,
|
||||
memberships,
|
||||
ownMembershipIdentity: ownMemberMock,
|
||||
roomId: "!room:example.org",
|
||||
slotId: "m.call#ROOM",
|
||||
});
|
||||
|
||||
it("returns preferred transport", async () => {
|
||||
// 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(() => {
|
||||
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),
|
||||
),
|
||||
};
|
||||
});
|
||||
|
||||
@@ -71,12 +90,19 @@ describe("LocalTransport", () => {
|
||||
fetchMock.reset();
|
||||
});
|
||||
|
||||
it("throws if config is missing", async () => {
|
||||
await expect(getLocalTransport(params())).rejects.toThrow(
|
||||
new MatrixRTCTransportMissingError("example.org"),
|
||||
);
|
||||
});
|
||||
|
||||
describe("local transport configuration mechanisms", () => {
|
||||
it("supports getting transport via application config", async () => {
|
||||
mockConfig({
|
||||
livekit: { livekit_service_url: "https://lk.example.org" },
|
||||
});
|
||||
|
||||
expect(await getLocalTransport(client)).toStrictEqual({
|
||||
expect(await getLocalTransport(params())).toStrictEqual({
|
||||
transport: {
|
||||
livekit_service_url: "https://lk.example.org",
|
||||
type: "livekit",
|
||||
@@ -91,7 +117,7 @@ describe("LocalTransport", () => {
|
||||
livekit_service_url: "https://lk.example.org",
|
||||
});
|
||||
|
||||
expect(await getLocalTransport(client)).toStrictEqual({
|
||||
expect(await getLocalTransport(params())).toStrictEqual({
|
||||
transport: {
|
||||
livekit_service_url: "https://lk.example.org",
|
||||
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 () => {
|
||||
client._unstable_getRTCTransports.mockResolvedValue([
|
||||
{ type: "livekit", livekit_service_url: "https://lk.example.org" },
|
||||
]);
|
||||
|
||||
expect(await getLocalTransport(client)).toStrictEqual({
|
||||
expect(await getLocalTransport(params())).toStrictEqual({
|
||||
transport: {
|
||||
livekit_service_url: "https://lk.example.org",
|
||||
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 () => {
|
||||
await expect(
|
||||
getLocalTransport({
|
||||
getDomain: () => "example.org",
|
||||
_unstable_getRTCTransports: async () => Promise.resolve([]),
|
||||
await expect(getLocalTransport(params())).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"),
|
||||
}),
|
||||
).rejects.toThrow(new MatrixRTCTransportMissingError("example.org"));
|
||||
);
|
||||
|
||||
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 { 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 { MatrixRTCTransportMissingError } from "../../../utils/errors.ts";
|
||||
import { customTransport as customTransportSetting } from "../../../settings/settings.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
|
||||
* 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(
|
||||
client: Pick<MatrixClient, "getDomain" | "_unstable_getRTCTransports">,
|
||||
): Promise<TransportLocator> {
|
||||
const serverName = client.getDomain();
|
||||
if (serverName === null) throw new Error("No server name");
|
||||
export async function getLocalTransport({
|
||||
client,
|
||||
memberships,
|
||||
ownMembershipIdentity,
|
||||
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({
|
||||
client,
|
||||
resolvedConfig: Config.get(),
|
||||
logger: rootLogger.getChild("[LocalTransport]"),
|
||||
logger,
|
||||
});
|
||||
|
||||
// Respect the user's custom transport, if set
|
||||
@@ -35,7 +85,64 @@ export async function getLocalTransport(
|
||||
customTransportSetting.value$.value ??
|
||||
(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 };
|
||||
}
|
||||
|
||||
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;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -83,6 +83,26 @@ describe("SessionBehaviors", () => {
|
||||
).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", () => {
|
||||
const dave = mockRtcMembership("@dave:example.org", "DEV222", {
|
||||
fociPreferred: [{ type: "something-else" } as Transport],
|
||||
|
||||
@@ -51,7 +51,9 @@ export const membershipsAndTransports$ = (
|
||||
transport: isUnstableLivekitTransport(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,
|
||||
};
|
||||
|
||||
@@ -111,6 +111,8 @@ describe("widget", () => {
|
||||
receiveToDevice: sendRecvToDevice,
|
||||
turnServers: false,
|
||||
rtcTransports: true,
|
||||
rtcLivekitGetToken: true,
|
||||
rtcLivekitDelegateDelayedLeave: true,
|
||||
sendDelayedEvents: true,
|
||||
updateDelayedEvents: true,
|
||||
sendSticky: true,
|
||||
|
||||
@@ -175,6 +175,8 @@ export const initializeWidget = (
|
||||
receiveToDevice: sendRecvToDevice,
|
||||
turnServers: false,
|
||||
rtcTransports: true,
|
||||
rtcLivekitGetToken: true,
|
||||
rtcLivekitDelegateDelayedLeave: true,
|
||||
sendDelayedEvents: true,
|
||||
updateDelayedEvents: true,
|
||||
sendSticky: true,
|
||||
|
||||
Reference in New Issue
Block a user