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
Port of edf1feae from johannes/rtc-remote-publish onto the SDK split,
where
the MatrixRTC logic lives in sdk/ rather than src/state/CallViewModel/.
Where the original lands:
- getLocalTransport's remote fallback: the LiveKit backend's new
findRemotePublishTransport (sdk/src/media-backend/livekit/
RemotePublishTransport.ts), which tries the candidates from the
js-sdk's
getRemoteLivekitPublishCandidates oldest first and keeps the first
whose
homeserver issues a token with the publish grant.
discoverLocalTransport
asks the backend for it after the host's fallback, with a snapshot of
the
session's memberships from RTCParticipation, and throws
MatrixRTCTransportMissingError where that finds nothing too. Transport
discovery stays backend neutral: the MSC4195 token check is the
backend's.
- LocalMember's "no delegation on a remote SFU" and the custom
transport's
server_name: LivekitBackend.prepareLocalTransport locates the
transport on
the homeserver it names (serverNameOfTransport) and skips the
delegation
probe where that is not our own.
- SessionBehaviors' membership.getTransportServerName: the same helper
in
LivekitBackend.mediaFor.
- decodeSFUJWTPayload, SFUJWTPayload.video.canPublish and the exported
isCsApiUnavailable: sdk/src/media-backend/livekit/auth/.
- DeveloperSettingsTab, the widget capabilities, the locale string and
the
js-sdk pin (9405287c, 44.0.0): applied as is. The error boundary
snapshots follow the string.
Tests: the LocalTransport cases go to RemotePublishTransport.test.ts and
LocalTransport.test.ts, the LocalMember remote case and the
SessionBehaviors
server-name case to LivekitBackend.test.ts, plus the memberships
snapshot in
RTCParticipation.test.ts.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Fable 5.1
parent
4b73ce68e3
commit
8638d4f159
+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
@@ -107,7 +107,7 @@
|
||||
"livekit-client": "^2.18.1",
|
||||
"lodash-es": "^4.17.21",
|
||||
"loglevel": "^1.9.1",
|
||||
"matrix-js-sdk": "github:matrix-org/matrix-js-sdk#develop",
|
||||
"matrix-js-sdk": "github:matrix-org/matrix-js-sdk#9405287c22797e9d69acd366c47bf412fa4384c8",
|
||||
"matrix-widget-api": "^1.18.0",
|
||||
"node-stdlib-browser": "^1.3.1",
|
||||
"normalize.css": "^8.0.1",
|
||||
|
||||
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
|
||||
|
||||
+11
-2
@@ -101,6 +101,8 @@ LiveKit participants.
|
||||
│ create ───▶│ │ │
|
||||
│ │ GET rtc/transports (session) ────────▶│ │
|
||||
│ │◀── first transport of the backend's type │
|
||||
│ │ (none: findRemotePublishTransport, │ │
|
||||
│ │ a member's SFU we may publish on) ─▶│ │
|
||||
│ │ prepareLocalTransport (backend): │ │
|
||||
│ │ publisher token: CS API get_token, │ │
|
||||
│ │ or OpenID token then /sfu/get ─────▶│ │
|
||||
@@ -537,6 +539,10 @@ type MemberMediaFields = Pick<RTCMember, "tracks$" | "encryptionError$">;
|
||||
|
||||
export interface MediaBackend {
|
||||
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. */
|
||||
prepareLocalTransport(
|
||||
transport: Transport,
|
||||
@@ -556,7 +562,9 @@ export interface MediaBackend {
|
||||
```
|
||||
|
||||
What stays on the client's side: discovery of the local transport from the
|
||||
homeserver's list (a homeserver endpoint, so not a backend's), the membership
|
||||
homeserver's list (a homeserver endpoint, so not a backend's), asking the
|
||||
backend for another member's transport only where that list and the host's
|
||||
fallback are empty, the membership
|
||||
state machine in `LocalMembership.ts` and its combination with
|
||||
`local.connectionState$` into `state$`, the choice of delayed leave timings,
|
||||
mapping a data packet's sender to a member, and the transport registry, which keys on the raw
|
||||
@@ -566,7 +574,8 @@ transport serialised with sorted keys and derives `connectedTransports$` and eac
|
||||
The backend's context carries no Matrix client and no membership list. A backend
|
||||
that needs the client takes a `Pick` of exactly what it calls in its own options
|
||||
(`createLivekitBackend(scope, context, { client, mediaQuality })`),
|
||||
and learns of remote members only through `mediaFor$`, registering each
|
||||
and learns of remote members only through `mediaFor$` and the one membership
|
||||
snapshot `findRemotePublishTransport` is handed, registering each
|
||||
membership's transport for as long as the member's scope lives. Media keys reach
|
||||
it as a stream, `mediaKeys$`, replayed on subscribe; the js-sdk session never
|
||||
does. A host supplies its own backend through `MatrixRTCClientOptions.backend`;
|
||||
|
||||
@@ -84,6 +84,7 @@ interface FakeBackend {
|
||||
connectionState$: BehaviorSubject<MediaConnectionState | Error>;
|
||||
data$: Subject<DataPacket>;
|
||||
setPublishing: Mock<(publish: boolean) => void>;
|
||||
findRemotePublishTransport: Mock<() => Promise<Transport | null>>;
|
||||
delegateDelayedLeave: Mock<(delayId: string) => Promise<void>>;
|
||||
mediaFor: Mock<(scope: ObservableScope, member: RTCMember) => MemberMedia>;
|
||||
}
|
||||
@@ -91,6 +92,7 @@ interface FakeBackend {
|
||||
function setup(
|
||||
options: Partial<RTCParticipationOptions> = {},
|
||||
slotOptions: { matrixRTCMode?: MatrixRTCMode } = {},
|
||||
homeserverTransports: Transport[] = [transport],
|
||||
): {
|
||||
scope: ObservableScope;
|
||||
participation: ReturnType<typeof createRTCParticipation>["participation"];
|
||||
@@ -112,7 +114,7 @@ function setup(
|
||||
getDeviceId: () => deviceId,
|
||||
getDomain: () => "example.org",
|
||||
getSyncState: () => SyncState.Syncing,
|
||||
_unstable_getRTCTransports: vi.fn().mockResolvedValue([transport]),
|
||||
_unstable_getRTCTransports: vi.fn().mockResolvedValue(homeserverTransports),
|
||||
} as unknown as MatrixClient;
|
||||
const members$ = new BehaviorSubject<RTCMember[]>([]);
|
||||
|
||||
@@ -124,6 +126,7 @@ function setup(
|
||||
),
|
||||
data$: new Subject<DataPacket>(),
|
||||
setPublishing: vi.fn(),
|
||||
findRemotePublishTransport: vi.fn(async () => Promise.resolve(null)),
|
||||
delegateDelayedLeave: vi.fn(async () => Promise.resolve()),
|
||||
mediaFor: vi.fn(() => ({
|
||||
tracks$: constant(null),
|
||||
@@ -136,6 +139,7 @@ function setup(
|
||||
): MediaBackend => {
|
||||
const backend: MediaBackend = {
|
||||
transportType: "livekit",
|
||||
findRemotePublishTransport: fake.findRemotePublishTransport,
|
||||
prepareLocalTransport: async () => fake.prepared.promise,
|
||||
delegateDelayedLeave: fake.delegateDelayedLeave,
|
||||
local: {
|
||||
@@ -220,6 +224,27 @@ describe("createRTCParticipation", () => {
|
||||
});
|
||||
});
|
||||
|
||||
it("publishes on another member's SFU where the homeserver advertises none", async () => {
|
||||
const bobsTransport: Transport = {
|
||||
type: "livekit",
|
||||
url: "wss://sfu.bob.com",
|
||||
server_name: "bob.com",
|
||||
};
|
||||
const { session, join, fake } = setup({}, {}, []);
|
||||
fake.findRemotePublishTransport.mockResolvedValue(bobsTransport);
|
||||
fake.prepared.resolve({ canDelegateDelayedLeave: false });
|
||||
await flushPromises();
|
||||
|
||||
expect(fake.findRemotePublishTransport).toHaveBeenCalledWith(
|
||||
session.memberships,
|
||||
);
|
||||
expect(join).toHaveBeenCalledWith(
|
||||
expect.anything(),
|
||||
[bobsTransport],
|
||||
expect.anything(),
|
||||
);
|
||||
});
|
||||
|
||||
it("publishes only while joined", async () => {
|
||||
const { participation, session, fake, scope } = setup();
|
||||
const ended = vi.fn();
|
||||
|
||||
@@ -131,12 +131,18 @@ export function createRTCParticipation(
|
||||
});
|
||||
|
||||
const preparedTransport$ = from(
|
||||
discoverLocalTransport(client, backend.transportType, options, logger).then(
|
||||
async (transport): Promise<PreparedTransport> => ({
|
||||
transport,
|
||||
...(await backend.prepareLocalTransport(transport)),
|
||||
}),
|
||||
),
|
||||
discoverLocalTransport(
|
||||
client,
|
||||
backend,
|
||||
// A snapshot: if nobody with a usable SFU is here yet, the user has to
|
||||
// come back later rather than wait in a half-working lobby
|
||||
jsSdkSession.memberships,
|
||||
options,
|
||||
logger,
|
||||
).then(async (transport): Promise<PreparedTransport> => ({
|
||||
transport,
|
||||
...(await backend.prepareLocalTransport(transport)),
|
||||
})),
|
||||
);
|
||||
|
||||
const localMembership = createLocalMembership$({
|
||||
|
||||
@@ -10,6 +10,7 @@ import { type Transport } from "matrix-js-sdk/lib/matrixrtc";
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
|
||||
import { MatrixRTCTransportMissingError } from "../errors";
|
||||
import { mockRtcMembership } from "../utils/test";
|
||||
import { discoverLocalTransport } from "./LocalTransport";
|
||||
|
||||
const homeserverTransport: Transport = {
|
||||
@@ -20,6 +21,12 @@ const hostTransport: Transport = {
|
||||
type: "livekit",
|
||||
livekit_service_url: "https://host.example.org",
|
||||
};
|
||||
const bobsTransport: Transport = {
|
||||
type: "livekit",
|
||||
url: "wss://sfu.bob.com",
|
||||
server_name: "bob.com",
|
||||
};
|
||||
const bob = mockRtcMembership("@bob:bob.com", "BOB");
|
||||
|
||||
function client(
|
||||
advertised: Transport[],
|
||||
@@ -30,6 +37,15 @@ function client(
|
||||
};
|
||||
}
|
||||
|
||||
function backend(
|
||||
remote: Transport | null = null,
|
||||
): Parameters<typeof discoverLocalTransport>[1] {
|
||||
return {
|
||||
transportType: "livekit",
|
||||
findRemotePublishTransport: vi.fn().mockResolvedValue(remote),
|
||||
};
|
||||
}
|
||||
|
||||
describe("discoverLocalTransport", () => {
|
||||
it("takes the transport the host names without asking the homeserver", async () => {
|
||||
const c = client([homeserverTransport]);
|
||||
@@ -37,7 +53,8 @@ describe("discoverLocalTransport", () => {
|
||||
await expect(
|
||||
discoverLocalTransport(
|
||||
c,
|
||||
"livekit",
|
||||
backend(),
|
||||
[],
|
||||
{ transport: hostTransport },
|
||||
logger,
|
||||
),
|
||||
@@ -49,7 +66,8 @@ describe("discoverLocalTransport", () => {
|
||||
await expect(
|
||||
discoverLocalTransport(
|
||||
client([homeserverTransport]),
|
||||
"livekit",
|
||||
backend(),
|
||||
[],
|
||||
{ fallbackTransport: hostTransport },
|
||||
logger,
|
||||
),
|
||||
@@ -60,15 +78,46 @@ describe("discoverLocalTransport", () => {
|
||||
await expect(
|
||||
discoverLocalTransport(
|
||||
client([{ type: "not_livekit" }]),
|
||||
"livekit",
|
||||
backend(),
|
||||
[],
|
||||
{ fallbackTransport: hostTransport },
|
||||
logger,
|
||||
),
|
||||
).resolves.toEqual(hostTransport);
|
||||
});
|
||||
|
||||
it("prefers a local transport over other members' SFUs", async () => {
|
||||
const b = backend(bobsTransport);
|
||||
|
||||
await expect(
|
||||
discoverLocalTransport(
|
||||
client([homeserverTransport]),
|
||||
b,
|
||||
[bob],
|
||||
{},
|
||||
logger,
|
||||
),
|
||||
).resolves.toEqual(homeserverTransport);
|
||||
expect(b.findRemotePublishTransport).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("publishes on another member's SFU where there is no local transport", async () => {
|
||||
const b = backend(bobsTransport);
|
||||
|
||||
await expect(
|
||||
discoverLocalTransport(client([]), b, [bob], {}, logger),
|
||||
).resolves.toEqual(bobsTransport);
|
||||
expect(b.findRemotePublishTransport).toHaveBeenCalledWith([bob]);
|
||||
});
|
||||
|
||||
it("fails naming the homeserver where there is no transport at all", async () => {
|
||||
const promise = discoverLocalTransport(client([]), "livekit", {}, logger);
|
||||
const promise = discoverLocalTransport(
|
||||
client([]),
|
||||
backend(),
|
||||
[bob],
|
||||
{},
|
||||
logger,
|
||||
);
|
||||
|
||||
await expect(promise).rejects.toBeInstanceOf(
|
||||
MatrixRTCTransportMissingError,
|
||||
|
||||
@@ -7,19 +7,27 @@ Please see LICENSE in the repository root for full details.
|
||||
|
||||
import { type MatrixClient } from "matrix-js-sdk";
|
||||
import { type Logger } from "matrix-js-sdk/lib/logger";
|
||||
import { type Transport } from "matrix-js-sdk/lib/matrixrtc";
|
||||
import {
|
||||
type CallMembership,
|
||||
type Transport,
|
||||
} from "matrix-js-sdk/lib/matrixrtc";
|
||||
|
||||
import { type RTCParticipationOptions } from "../api";
|
||||
import { MatrixRTCTransportMissingError } from "../errors";
|
||||
import { type MediaBackend } from "../media-backend/api";
|
||||
import { TransportAutoDiscovery } from "./TransportAutoDiscovery";
|
||||
|
||||
/**
|
||||
* The transport the local member advertises: the one the host names, else the
|
||||
* homeserver's first of the type the backend serves, else the host's fallback.
|
||||
* homeserver's first of the type the backend serves, else the host's fallback,
|
||||
* else another member's that the backend can publish on.
|
||||
*
|
||||
* @throws {MatrixRTCTransportMissingError} where none of these exists.
|
||||
*/
|
||||
export async function discoverLocalTransport(
|
||||
client: Pick<MatrixClient, "getDomain" | "_unstable_getRTCTransports">,
|
||||
transportType: string,
|
||||
backend: Pick<MediaBackend, "transportType" | "findRemotePublishTransport">,
|
||||
memberships: CallMembership[],
|
||||
{
|
||||
transport: override,
|
||||
fallbackTransport,
|
||||
@@ -30,10 +38,11 @@ export async function discoverLocalTransport(
|
||||
override ??
|
||||
(await new TransportAutoDiscovery({
|
||||
client,
|
||||
transportType,
|
||||
transportType: backend.transportType,
|
||||
fallbackTransport,
|
||||
logger,
|
||||
}).discoverPreferredTransport());
|
||||
}).discoverPreferredTransport()) ??
|
||||
(await backend.findRemotePublishTransport(memberships));
|
||||
if (transport === null)
|
||||
throw new MatrixRTCTransportMissingError(client.getDomain() ?? "");
|
||||
return transport;
|
||||
|
||||
@@ -13,7 +13,10 @@ Please see LICENSE in the repository root for full details.
|
||||
*/
|
||||
|
||||
import { type Logger } from "matrix-js-sdk/lib/logger";
|
||||
import { type Transport } from "matrix-js-sdk/lib/matrixrtc";
|
||||
import {
|
||||
type CallMembership,
|
||||
type Transport,
|
||||
} from "matrix-js-sdk/lib/matrixrtc";
|
||||
import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager";
|
||||
import { type Observable } from "rxjs";
|
||||
|
||||
@@ -116,6 +119,17 @@ export interface MediaBackend {
|
||||
/** The membership transport type this backend serves: "livekit". */
|
||||
readonly transportType: string;
|
||||
|
||||
/**
|
||||
* A transport to publish on where neither the homeserver nor the host
|
||||
* provides one of the backend's type: for LiveKit, the SFU of another
|
||||
* member whose homeserver lets us publish on it (MSC4195). Resolves null
|
||||
* where no member's transport will do. Called at most once per session,
|
||||
* with the memberships at that moment, before `prepareLocalTransport`.
|
||||
*/
|
||||
findRemotePublishTransport(
|
||||
memberships: CallMembership[],
|
||||
): Promise<Transport | null>;
|
||||
|
||||
/**
|
||||
* 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
|
||||
|
||||
@@ -25,6 +25,7 @@ import { defaultSessionTimings } from "../../config";
|
||||
import {
|
||||
flushPromises,
|
||||
mockEmitter,
|
||||
mockRtcMembership,
|
||||
ownMemberMock,
|
||||
testScope,
|
||||
} from "../../utils/test";
|
||||
@@ -258,6 +259,52 @@ describe("createLivekitBackend", () => {
|
||||
},
|
||||
);
|
||||
|
||||
it("prepares a remote SFU's transport on its homeserver, where delegation is not possible", async () => {
|
||||
// Our own homeserver would delegate, so only the SFU being remote can
|
||||
// explain delegation being skipped
|
||||
fetchMock.post(homeserverEndpoint, 401);
|
||||
const remoteTransport: UnstableLivekitTransport = {
|
||||
type: "livekit",
|
||||
url: "wss://sfu.sample.com",
|
||||
server_name: "sample.com",
|
||||
};
|
||||
|
||||
const capabilities = await backend().prepareLocalTransport(remoteTransport);
|
||||
|
||||
expect(capabilities).toEqual({ canDelegateDelayedLeave: false });
|
||||
expect(connections[0].connection.role).toBe("publisher");
|
||||
expect(connections[0].connection.serverName).toBe("sample.com");
|
||||
});
|
||||
|
||||
it("finds another member's SFU to publish on through the homeserver", async () => {
|
||||
client._unstable_getLivekitToken.mockResolvedValue({
|
||||
jwt: [
|
||||
{},
|
||||
{ sub: "identity", video: { room: "room", canPublish: true } },
|
||||
{},
|
||||
]
|
||||
.map((d) => global.btoa(JSON.stringify(d)))
|
||||
.join("."),
|
||||
});
|
||||
const carl = mockRtcMembership("@carl:sample.com", "CARL", {
|
||||
fociPreferred: [{ type: "livekit", url: "wss://sfu.sample.com" }],
|
||||
});
|
||||
|
||||
await expect(backend().findRemotePublishTransport([carl])).resolves.toEqual(
|
||||
{
|
||||
type: "livekit",
|
||||
url: "wss://sfu.sample.com",
|
||||
server_name: "sample.com",
|
||||
},
|
||||
);
|
||||
expect(client._unstable_getLivekitToken).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
url: "wss://sfu.sample.com",
|
||||
server_name: "sample.com",
|
||||
}),
|
||||
);
|
||||
});
|
||||
|
||||
it("waits for the publisher connection to hold its token before the join", async () => {
|
||||
fetchMock.post(homeserverEndpoint, 404);
|
||||
fetchMock.post(`${exampleServiceUrl}/delegate_delayed_leave`, 404);
|
||||
@@ -488,6 +535,27 @@ describe("createLivekitBackend", () => {
|
||||
expect(remote?.connection.transport).toEqual(carlsTransport);
|
||||
expect(remote?.connection.serverName).toBe("sample.com");
|
||||
});
|
||||
|
||||
it("connects to a remote publisher's transport on the SFU's homeserver", async () => {
|
||||
const b = await prepared();
|
||||
// Dave has no SFU of his own and publishes on sample.com's
|
||||
const remoteTransport: UnstableLivekitTransport = {
|
||||
type: "livekit",
|
||||
url: "wss://sfu.sample.com",
|
||||
server_name: "sample.com",
|
||||
};
|
||||
b.mediaFor(testScope(), {
|
||||
...bob(remoteTransport, ["dave-hashed"]),
|
||||
userId: "@dave:example.org",
|
||||
});
|
||||
await flushPromises();
|
||||
|
||||
const remote = connections.find(
|
||||
(c) => c.connection.role === "subscriber",
|
||||
);
|
||||
expect(remote?.connection.transport).toEqual(remoteTransport);
|
||||
expect(remote?.connection.serverName).toBe("sample.com");
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
@@ -8,6 +8,7 @@ Please see LICENSE in the repository root for full details.
|
||||
import { type MatrixClient } from "matrix-js-sdk";
|
||||
import { type Logger } from "matrix-js-sdk/lib/logger";
|
||||
import {
|
||||
type CallMembership,
|
||||
isUnstableLivekitTransport,
|
||||
type Transport,
|
||||
} from "matrix-js-sdk/lib/matrixrtc";
|
||||
@@ -34,7 +35,7 @@ import {
|
||||
trackEpoch,
|
||||
} from "../../reactive/ObservableScope";
|
||||
import { mapScoped } from "../../utils/mapScoped";
|
||||
import { serverNameOf } from "../../utils/serverName";
|
||||
import { serverNameOf, serverNameOfTransport } from "../../utils/serverName";
|
||||
import {
|
||||
type BackendConnection,
|
||||
type MediaBackend,
|
||||
@@ -62,6 +63,7 @@ import { createDataChannel$ } from "./DataChannel";
|
||||
import { createKeyProvider } from "./KeyProvider";
|
||||
import { createLivekitMemberMedia } from "./LivekitMemberMedia";
|
||||
import { createLivekitLocalMedia } from "./LocalMedia";
|
||||
import { findRemotePublishTransport } from "./RemotePublishTransport";
|
||||
|
||||
export interface LivekitBackendOptions {
|
||||
/**
|
||||
@@ -146,26 +148,50 @@ export function createLivekitBackend(
|
||||
logger,
|
||||
);
|
||||
|
||||
const ownServerName = serverNameOf(ownMembershipIdentity.userId);
|
||||
|
||||
const remotePublishTransport = async (
|
||||
memberships: CallMembership[],
|
||||
): Promise<Transport | null> =>
|
||||
findRemotePublishTransport({
|
||||
client,
|
||||
memberships,
|
||||
ownMembershipIdentity,
|
||||
roomId,
|
||||
slotId,
|
||||
logger,
|
||||
});
|
||||
|
||||
const prepareLocalTransport = async (
|
||||
transport: Transport,
|
||||
): Promise<TransportCapabilities> => {
|
||||
if (!isUnstableLivekitTransport(transport))
|
||||
throw new MatrixRTCError(`Not a LiveKit transport: ${transport.type}`);
|
||||
const locator: TransportLocator = {
|
||||
const serverName = serverNameOfTransport(
|
||||
transport,
|
||||
serverName: serverNameOf(ownMembershipIdentity.userId),
|
||||
};
|
||||
// Delegation goes through whichever service authenticates us as a
|
||||
// publisher, so only that service's support counts
|
||||
const flow = chooseAuthFlow(transport, "publisher");
|
||||
const canDelegate =
|
||||
flow.kind === "openID"
|
||||
? supportsDelegation(
|
||||
`${flow.serviceUrl}/delegate_delayed_leave`,
|
||||
`transport ${flow.serviceUrl}`,
|
||||
logger,
|
||||
)
|
||||
: homeserverSupportsDelegation;
|
||||
ownMembershipIdentity.userId,
|
||||
);
|
||||
const locator: TransportLocator = { transport, serverName };
|
||||
let canDelegate: Promise<boolean>;
|
||||
if (serverName !== ownServerName) {
|
||||
// Our homeserver only takes over leaves for its own SFUs
|
||||
logger.info(
|
||||
`Publishing on a remote SFU of ${serverName}, delegation is not possible`,
|
||||
);
|
||||
canDelegate = Promise.resolve(false);
|
||||
} else {
|
||||
// Delegation goes through whichever service authenticates us as a
|
||||
// publisher, so only that service's support counts
|
||||
const flow = chooseAuthFlow(transport, "publisher");
|
||||
canDelegate =
|
||||
flow.kind === "openID"
|
||||
? supportsDelegation(
|
||||
`${flow.serviceUrl}/delegate_delayed_leave`,
|
||||
`transport ${flow.serviceUrl}`,
|
||||
logger,
|
||||
)
|
||||
: homeserverSupportsDelegation;
|
||||
}
|
||||
localTransport$.next(locator);
|
||||
// The membership is only sent once the publisher connection holds its
|
||||
// token, so that the LiveKit room exists before anyone subscribes to it:
|
||||
@@ -204,7 +230,10 @@ export function createLivekitBackend(
|
||||
map((transport): TransportLocator | undefined => {
|
||||
const raw = transport?.raw;
|
||||
return isUnstableLivekitTransport(raw)
|
||||
? { transport: raw, serverName: serverNameOf(member.userId) }
|
||||
? {
|
||||
transport: raw,
|
||||
serverName: serverNameOfTransport(raw, member.userId),
|
||||
}
|
||||
: undefined;
|
||||
}),
|
||||
distinctUntilChanged((a, b) => (a && keyFor(a)) === (b && keyFor(b))),
|
||||
@@ -257,6 +286,7 @@ export function createLivekitBackend(
|
||||
|
||||
return {
|
||||
transportType: "livekit",
|
||||
findRemotePublishTransport: remotePublishTransport,
|
||||
prepareLocalTransport,
|
||||
delegateDelayedLeave: delegateLeave,
|
||||
local: localMedia_,
|
||||
|
||||
@@ -0,0 +1,149 @@
|
||||
/*
|
||||
Copyright 2026 Element Creations Ltd.
|
||||
|
||||
SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
|
||||
Please see LICENSE in the repository root for full details.
|
||||
*/
|
||||
|
||||
import { MatrixError } from "matrix-js-sdk";
|
||||
import { logger } from "matrix-js-sdk/lib/logger";
|
||||
import { type CallMembership } from "matrix-js-sdk/lib/matrixrtc";
|
||||
import { beforeEach, describe, expect, it, type Mocked, vi } from "vitest";
|
||||
|
||||
import { mockRtcMembership, ownMemberMock } from "../../utils/test";
|
||||
import { type ClientGetTokenParts } from "./auth/csApi";
|
||||
import { findRemotePublishTransport } from "./RemotePublishTransport";
|
||||
|
||||
/** A JWT whose payload states whether the issuing server lets us publish. */
|
||||
function jwtWithPublishGrant(canPublish: boolean): string {
|
||||
return [
|
||||
{}, // header
|
||||
{ sub: "identity", video: { room: "room", canPublish } }, // payload
|
||||
{}, // signature
|
||||
]
|
||||
.map((d) => global.btoa(JSON.stringify(d)))
|
||||
.join(".");
|
||||
}
|
||||
|
||||
const bobSfu = { type: "livekit", url: "wss://sfu.bob.com" };
|
||||
const carolSfu = { type: "livekit", url: "wss://sfu.carol.com" };
|
||||
|
||||
// Bob is the oldest member, Carol joined after him
|
||||
const bob = mockRtcMembership("@bob:bob.com", "BOB", {
|
||||
fociPreferred: [bobSfu],
|
||||
membership: { created_ts: 100 },
|
||||
});
|
||||
const carol = mockRtcMembership("@carol:carol.com", "CAROL", {
|
||||
fociPreferred: [carolSfu],
|
||||
membership: { created_ts: 200 },
|
||||
});
|
||||
const legacyDave = mockRtcMembership("@dave:dave.com", "DAVE", {
|
||||
fociPreferred: [
|
||||
{ type: "livekit", livekit_service_url: "https://jwt.dave.com" },
|
||||
],
|
||||
membership: { created_ts: 50 },
|
||||
});
|
||||
|
||||
const csApiMissing = new MatrixError(
|
||||
{ errcode: "M_UNRECOGNIZED", error: "" },
|
||||
404,
|
||||
);
|
||||
|
||||
describe("findRemotePublishTransport", () => {
|
||||
let client: Mocked<ClientGetTokenParts>;
|
||||
const find = async (
|
||||
memberships: CallMembership[],
|
||||
): ReturnType<typeof findRemotePublishTransport> =>
|
||||
findRemotePublishTransport({
|
||||
client,
|
||||
memberships,
|
||||
ownMembershipIdentity: ownMemberMock,
|
||||
roomId: "!room:example.org",
|
||||
slotId: "m.call#ROOM",
|
||||
logger,
|
||||
});
|
||||
|
||||
beforeEach(() => {
|
||||
client = {
|
||||
_unstable_getLivekitToken: vi
|
||||
.fn()
|
||||
.mockResolvedValue({ jwt: jwtWithPublishGrant(true) }),
|
||||
};
|
||||
});
|
||||
|
||||
it("publishes on the oldest member's SFU with its server_name", async () => {
|
||||
await expect(find([carol, bob])).resolves.toEqual({
|
||||
type: "livekit",
|
||||
url: "wss://sfu.bob.com",
|
||||
server_name: "bob.com",
|
||||
});
|
||||
expect(client._unstable_getLivekitToken).toHaveBeenCalledTimes(1);
|
||||
expect(client._unstable_getLivekitToken).toHaveBeenCalledWith({
|
||||
url: "wss://sfu.bob.com",
|
||||
server_name: "bob.com",
|
||||
room_id: "!room:example.org",
|
||||
slot_id: "m.call#ROOM",
|
||||
member_id: ownMemberMock.memberId,
|
||||
});
|
||||
});
|
||||
|
||||
it("ignores members publishing through a legacy JWT service", async () => {
|
||||
await expect(find([legacyDave, bob])).resolves.toMatchObject({
|
||||
server_name: "bob.com",
|
||||
});
|
||||
expect(client._unstable_getLivekitToken).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("moves on when a homeserver does not let us publish", async () => {
|
||||
client._unstable_getLivekitToken.mockImplementation(
|
||||
async ({ server_name }) =>
|
||||
Promise.resolve({
|
||||
jwt: jwtWithPublishGrant(server_name === "carol.com"),
|
||||
}),
|
||||
);
|
||||
|
||||
await expect(find([bob, carol])).resolves.toMatchObject({
|
||||
server_name: "carol.com",
|
||||
});
|
||||
expect(client._unstable_getLivekitToken).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it("moves on when a homeserver refuses to issue a token", async () => {
|
||||
client._unstable_getLivekitToken.mockImplementation(
|
||||
async ({ server_name }) =>
|
||||
server_name === "bob.com"
|
||||
? Promise.reject(
|
||||
new MatrixError(
|
||||
{ errcode: "M_FORBIDDEN", error: "Not joined" },
|
||||
403,
|
||||
),
|
||||
)
|
||||
: Promise.resolve({ jwt: jwtWithPublishGrant(true) }),
|
||||
);
|
||||
|
||||
await expect(find([bob, carol])).resolves.toMatchObject({
|
||||
server_name: "carol.com",
|
||||
});
|
||||
});
|
||||
|
||||
it("finds nothing once all candidates are exhausted", async () => {
|
||||
client._unstable_getLivekitToken.mockResolvedValue({
|
||||
jwt: jwtWithPublishGrant(false),
|
||||
});
|
||||
|
||||
await expect(find([bob, carol])).resolves.toBeNull();
|
||||
expect(client._unstable_getLivekitToken).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it("gives up immediately if our homeserver lacks the CS API", async () => {
|
||||
client._unstable_getLivekitToken.mockRejectedValue(csApiMissing);
|
||||
|
||||
await expect(find([bob, carol])).resolves.toBeNull();
|
||||
expect(client._unstable_getLivekitToken).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("finds nothing where there are no other members", async () => {
|
||||
await expect(find([])).resolves.toBeNull();
|
||||
expect(client._unstable_getLivekitToken).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,90 @@
|
||||
/*
|
||||
Copyright 2026 Element Creations Ltd.
|
||||
|
||||
SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
|
||||
Please see LICENSE in the repository root for full details.
|
||||
*/
|
||||
|
||||
import { type Logger } from "matrix-js-sdk/lib/logger";
|
||||
import {
|
||||
type CallMembership,
|
||||
getRemoteLivekitPublishCandidates,
|
||||
type LivekitPublishCandidate,
|
||||
type Transport,
|
||||
} from "matrix-js-sdk/lib/matrixrtc";
|
||||
import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager";
|
||||
|
||||
import { isCsApiUnavailable } from "./auth";
|
||||
import * as csApi from "./auth/csApi";
|
||||
import { decodeSFUJWTPayload } from "./auth/types";
|
||||
|
||||
export interface RemotePublishTransportParams {
|
||||
client: csApi.ClientGetTokenParts;
|
||||
/** The members whose SFUs are the candidates. */
|
||||
memberships: CallMembership[];
|
||||
ownMembershipIdentity: CallMembershipIdentityParts;
|
||||
roomId: string;
|
||||
slotId: string;
|
||||
logger: Logger;
|
||||
}
|
||||
|
||||
/**
|
||||
* The SFU of another member that its homeserver lets us publish on, as the
|
||||
* transport to advertise, or null where no member's SFU will do. Candidates
|
||||
* are tried oldest membership first, and each one is checked by asking for a
|
||||
* publisher token through our homeserver (MSC4195): a homeserver may restrict
|
||||
* remote publishers, which shows as a token without the publish grant.
|
||||
*/
|
||||
export async function findRemotePublishTransport({
|
||||
client,
|
||||
memberships,
|
||||
ownMembershipIdentity,
|
||||
roomId,
|
||||
slotId,
|
||||
logger,
|
||||
}: RemotePublishTransportParams): Promise<Transport | null> {
|
||||
logger.info(
|
||||
"No local transport available, looking for another member's SFU to publish on",
|
||||
);
|
||||
const candidates = getRemoteLivekitPublishCandidates(
|
||||
memberships,
|
||||
ownMembershipIdentity,
|
||||
);
|
||||
for (const candidate of candidates) {
|
||||
const outcome = await tryCandidate(candidate);
|
||||
if (outcome === "usable") return candidate.transport;
|
||||
if (outcome === "noCsApi") return null;
|
||||
}
|
||||
return null;
|
||||
|
||||
/** "noCsApi" ends the search: without the endpoint no candidate can work. */
|
||||
async function tryCandidate({
|
||||
url,
|
||||
serverName,
|
||||
}: LivekitPublishCandidate): Promise<"usable" | "unusable" | "noCsApi"> {
|
||||
try {
|
||||
const { jwt } = await csApi.getSFUConfig({
|
||||
client,
|
||||
membership: ownMembershipIdentity,
|
||||
url,
|
||||
serverName,
|
||||
roomId,
|
||||
slotId,
|
||||
});
|
||||
if (decodeSFUJWTPayload(jwt).video.canPublish) {
|
||||
logger.info(`Publishing on remote SFU ${url} of ${serverName}`);
|
||||
return "usable";
|
||||
}
|
||||
logger.info(`${serverName} does not let us publish on ${url}, skipping`);
|
||||
} catch (e) {
|
||||
if (isCsApiUnavailable(e)) {
|
||||
logger.warn(
|
||||
`Cannot use the MSC4195 get_token endpoint (${e}), so no remote SFU can be used`,
|
||||
);
|
||||
return "noCsApi";
|
||||
}
|
||||
logger.warn(`Failed to get a token for ${url} of ${serverName}:`, e);
|
||||
}
|
||||
return "unusable";
|
||||
}
|
||||
}
|
||||
@@ -48,7 +48,7 @@ function isEndpointUnsupported(e: unknown): boolean {
|
||||
* (MSC4533), which the widget API reports as a plain error. A connection
|
||||
* error is neither: the endpoint may well exist.
|
||||
*/
|
||||
function isCsApiUnavailable(e: unknown): boolean {
|
||||
export function isCsApiUnavailable(e: unknown): boolean {
|
||||
if (isEndpointUnsupported(e)) return true;
|
||||
return !(e instanceof MatrixError) && !(e instanceof ConnectionError);
|
||||
}
|
||||
|
||||
@@ -34,11 +34,13 @@ export interface SFUConfig {
|
||||
}
|
||||
|
||||
/** The parts of the JWT payload the backend reads. */
|
||||
interface SFUJWTPayload {
|
||||
export interface SFUJWTPayload {
|
||||
/** The subject: the participant identity the token carries. */
|
||||
sub: string;
|
||||
video: {
|
||||
room: string;
|
||||
/** Whether the token lets us publish; a homeserver may deny this to remote publishers. */
|
||||
canPublish?: boolean;
|
||||
};
|
||||
}
|
||||
|
||||
@@ -47,8 +49,7 @@ export function extractFullConfigFromToken(sfuConfig: {
|
||||
url: string;
|
||||
jwt: string;
|
||||
}): SFUConfig {
|
||||
const [, payloadStr] = sfuConfig.jwt.split(".");
|
||||
const payload = JSON.parse(global.atob(payloadStr)) as SFUJWTPayload;
|
||||
const payload = decodeSFUJWTPayload(sfuConfig.jwt);
|
||||
return {
|
||||
jwt: sfuConfig.jwt,
|
||||
url: sfuConfig.url,
|
||||
@@ -56,3 +57,9 @@ export function extractFullConfigFromToken(sfuConfig: {
|
||||
livekitIdentity: payload.sub,
|
||||
};
|
||||
}
|
||||
|
||||
/** Decodes the payload of an SFU JWT, without verifying its signature. */
|
||||
export function decodeSFUJWTPayload(jwt: string): SFUJWTPayload {
|
||||
const [, payloadStr] = jwt.split(".");
|
||||
return JSON.parse(global.atob(payloadStr)) as SFUJWTPayload;
|
||||
}
|
||||
|
||||
@@ -5,7 +5,23 @@ SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
|
||||
Please see LICENSE in the repository root for full details.
|
||||
*/
|
||||
|
||||
import { type Transport } from "matrix-js-sdk/lib/matrixrtc";
|
||||
|
||||
/** The homeserver a Matrix user id belongs to: everything after the first colon. */
|
||||
export function serverNameOf(userId: string): string {
|
||||
return userId.replace(/^.*?:/, "");
|
||||
}
|
||||
|
||||
/**
|
||||
* The homeserver that operates a transport: the one the transport names, else
|
||||
* the one of the member advertising it. A member publishing on another
|
||||
* homeserver's SFU (MSC4195) names that homeserver in `server_name`.
|
||||
*/
|
||||
export function serverNameOfTransport(
|
||||
transport: Transport,
|
||||
userId: string,
|
||||
): string {
|
||||
return typeof transport.server_name === "string"
|
||||
? transport.server_name
|
||||
: serverNameOf(userId);
|
||||
}
|
||||
|
||||
@@ -1228,7 +1228,7 @@ exports[`should have a close button when the host can dismiss us 1`] = `
|
||||
Call is not supported
|
||||
</h1>
|
||||
<p>
|
||||
The server is not configured to work with Element Call. Please contact your server admin (Domain: example.com, Error Code: MISSING_MATRIX_RTC_TRANSPORT).
|
||||
The server is not configured to work with Element Call and no other call participant provides a usable transport. Please contact your server admin (Domain: example.com, Error Code: MISSING_MATRIX_RTC_TRANSPORT).
|
||||
</p>
|
||||
<button
|
||||
class="_button_1nw83_8"
|
||||
@@ -1381,7 +1381,7 @@ exports[`should render the error page with link back to home 1`] = `
|
||||
Call is not supported
|
||||
</h1>
|
||||
<p>
|
||||
The server is not configured to work with Element Call. Please contact your server admin (Domain: example.com, Error Code: MISSING_MATRIX_RTC_TRANSPORT).
|
||||
The server is not configured to work with Element Call and no other call participant provides a usable transport. Please contact your server admin (Domain: example.com, Error Code: MISSING_MATRIX_RTC_TRANSPORT).
|
||||
</p>
|
||||
<button
|
||||
class="_button_1nw83_8"
|
||||
@@ -1534,7 +1534,7 @@ exports[`should report correct error for 'Call is not supported' 1`] = `
|
||||
Call is not supported
|
||||
</h1>
|
||||
<p>
|
||||
The server is not configured to work with Element Call. Please contact your server admin (Domain: example.com, Error Code: MISSING_MATRIX_RTC_TRANSPORT).
|
||||
The server is not configured to work with Element Call and no other call participant provides a usable transport. Please contact your server admin (Domain: example.com, Error Code: MISSING_MATRIX_RTC_TRANSPORT).
|
||||
</p>
|
||||
<button
|
||||
class="_button_1nw83_8"
|
||||
|
||||
@@ -517,7 +517,7 @@ export const DeveloperSettingsTab: FC<Props> = ({
|
||||
client,
|
||||
membership: { userId, deviceId, memberId: "" },
|
||||
transport,
|
||||
serverName,
|
||||
serverName: transport.server_name ?? serverName,
|
||||
roomId,
|
||||
slotId: "org.example.fake-slot#room",
|
||||
role: "publisher",
|
||||
|
||||
@@ -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