Compare commits

...
Author SHA1 Message Date
Johannes Marbach 8be229a438 Merge pull request #4354 from element-hq/toger5/johennes/rtc-remote-publish
Fall back to publishing on a remote SFU if possible
2026-10-15 16:27:36 +02:00
Timo K. ef7dce7b07 Update url-transport.test.ts 2026-10-15 16:10:02 +02:00
Timo K. 2de5ee2df5 Update test to consider EWs support for cs-api 2026-10-15 12:53:37 +02:00
Timo K. a427206792 Only return null if there is no endpoint available. 2026-10-15 12:42:36 +02:00
Timo K. 1913a099ab bump EW to inlcude the new widget capabilities 2026-10-15 11:28:45 +02:00
TimoandJohannes Marbach 93b893527e Update sdk/src/media-backend/livekit/RemotePublishTransport.ts
Co-authored-by: Johannes Marbach <n0-0ne+github@mailbox.org>
2026-10-15 09:28:40 +02:00
Timo K. 2b26fd472f fmt 2026-10-14 19:03:22 +02:00
Timo K. 281fff0e9f array -> prop|null in discoverLocalTransport 2026-10-14 19:01:14 +02:00
Johannes Marbach bc557dfb5d Revert "Temporarily enable full output to debug the Claude action"
This reverts commit 4827ab5fdc.
2026-10-14 17:55:35 +02:00
Johannes Marbach 4827ab5fdc Temporarily enable full output to debug the Claude action 2026-10-14 17:50:20 +02:00
Timo K. 7af6525f01 fix lints 2026-10-14 16:44:31 +02:00
Timo K. 3b763d87f2 make the transport discovery lk edgecase a EC exclusive workaround
Dont add it to the sdk.
2026-10-14 16:40:27 +02:00
Timo K.andClaude Fable 5.1 8638d4f159 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>
2026-10-14 16:09:11 +02:00
21 changed files with 523 additions and 53 deletions
+2 -2
View File
@@ -19,7 +19,7 @@ services:
- ./backend/playwright_homeserver-othersite.yaml:/data/cfg/homeserver.yaml:Z
element-web:
# Pin to a SHA so that upstream cannot break our tests. Renovate handles regular updates.
image: ghcr.io/element-hq/element-web:develop@sha256:e464b2aca73e248195b24b8c6f62cd331100302284d60cc735405807fb2730be
image: ghcr.io/element-hq/element-web:develop@sha256:b2a5a30afb4f583a7560448b005d8332bee29672f15826e4a995b393058f0291
element-web-1:
# Pin to a SHA so that upstream cannot break our tests. Renovate handles regular updates.
image: ghcr.io/element-hq/element-web:develop@sha256:e464b2aca73e248195b24b8c6f62cd331100302284d60cc735405807fb2730be
image: ghcr.io/element-hq/element-web:develop@sha256:b2a5a30afb4f583a7560448b005d8332bee29672f15826e4a995b393058f0291
+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.",
"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
View File
@@ -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#3f05dc4d104a3dfc753c5e3549a18c74fab4b076",
"matrix-widget-api": "^1.18.0",
"node-stdlib-browser": "^1.3.1",
"normalize.css": "^8.0.1",
+2 -6
View File
@@ -66,13 +66,9 @@ widgetTest(
await TestHelpers.expectVisibleVideoCount(frame, 2);
}
// Element Web does not expose the MSC4195 endpoints to widgets yet, so
// neither side reached the CS API. The host published through the JWT
// service's legacy endpoint and subscribed to the guest through its
// default one; the guest published the same way and fell back to the
// default endpoint for the host's transport.
// Element Web used for Playwright does support MSC4195 for widgets.
expect(hostFlows).toEqual(new Set(["service-legacy", "service-default"]));
expect(guestFlows).toEqual(new Set(["service-legacy", "service-default"]));
expect(guestFlows).toEqual(new Set(["service-legacy", "cs-api"]));
},
);
+6 -6
View File
@@ -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#3f05dc4d104a3dfc753c5e3549a18c74fab4b076
version: https://codeload.github.com/matrix-org/matrix-js-sdk/tar.gz/3f05dc4d104a3dfc753c5e3549a18c74fab4b076
matrix-widget-api:
specifier: ^1.18.0
version: 1.20.0
@@ -4673,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/3f05dc4d104a3dfc753c5e3549a18c74fab4b076:
resolution: {gitHosted: true, integrity: sha512-Ly8eS7fNNKDo9KmPvTkE49KvD8iMUmrVWJMNERNuxc8UksMLMG36NLz3fs8HytdwrB7AKPfyVo392VZsxILDIg==, tarball: https://codeload.github.com/matrix-org/matrix-js-sdk/tar.gz/3f05dc4d104a3dfc753c5e3549a18c74fab4b076}
version: 44.0.0
engines: {node: '>=22.0.0'}
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/3f05dc4d104a3dfc753c5e3549a18c74fab4b076:
dependencies:
'@babel/runtime': 8.0.0
'@matrix-org/matrix-sdk-crypto-wasm': 18.9.0
+6 -1
View File
@@ -101,6 +101,8 @@ LiveKit participants.
│ create ───▶│ │ │
│ │ GET rtc/transports (session) ────────▶│ │
│ │◀── first transport of the backend's type │
│ │ (none, LiveKit only: a member's SFU │ │
│ │ that lets us publish, get_token) ─▶│ │
│ │ prepareLocalTransport (backend): │ │
│ │ publisher token: CS API get_token, │ │
│ │ or OpenID token then /sfu/get ─────▶│ │
@@ -556,7 +558,10 @@ export interface MediaBackend {
```
What stays on the client's side: discovery of the local transport from the
homeserver's list (a homeserver endpoint, so not a backend's), the membership
homeserver's list (a homeserver endpoint, so not a backend's) with, for the
LiveKit backend only, a last resort of another member's SFU that lets us
publish (`findRemotePublishLivekitTransport`, wired in by `RTCParticipation` rather
than by the backend interface, since it is an MSC4195 matter), the membership
state machine in `LocalMembership.ts` and its combination with
`local.connectionState$` into `state$`, the choice of delayed leave timings,
mapping a data packet's sender to a member, and the transport registry, which keys on the raw
+51 -3
View File
@@ -8,6 +8,7 @@ Please see LICENSE in the repository root for full details.
import { type MatrixClient, type Room, SyncState } from "matrix-js-sdk";
import { logger } from "matrix-js-sdk/lib/logger";
import {
type CallMembership,
MatrixRTCSessionEvent,
MembershipManagerEvent,
type Transport,
@@ -38,13 +39,26 @@ import { constant } from "./reactive/Behavior";
import { ObservableScope } from "./reactive/ObservableScope";
import { createRTCParticipation } from "./RTCParticipation";
import { transportId } from "./matrixrtc/Transports";
import { flushPromises, mockEmitter, MockRTCSession } from "./utils/test";
import {
flushPromises,
mockEmitter,
mockRtcMembership,
MockRTCSession,
} from "./utils/test";
const transport: Transport = {
type: "livekit",
livekit_service_url: "https://lk.example.org",
};
const userId = "@alice:example.org";
/** A token with the publish grant, as a homeserver issues it for its own SFU. */
const publisherJWT = [
{},
{ sub: "identity", video: { room: "room", canPublish: true } },
{},
]
.map((d) => global.btoa(JSON.stringify(d)))
.join(".");
const deviceId = "AAAA";
const legacyMemberId = `${userId}:${deviceId}`;
@@ -91,6 +105,13 @@ interface FakeBackend {
function setup(
options: Partial<RTCParticipationOptions> = {},
slotOptions: { matrixRTCMode?: MatrixRTCMode } = {},
{
homeserverTransports = [transport],
memberships = [],
}: {
homeserverTransports?: Transport[];
memberships?: CallMembership[];
} = {},
): {
scope: ObservableScope;
participation: ReturnType<typeof createRTCParticipation>["participation"];
@@ -103,7 +124,7 @@ function setup(
} {
const scope = new ObservableScope();
const room = { roomId: "!room:example.org" } as Room;
const session = new MockRTCSession(room);
const session = new MockRTCSession(room, memberships);
const mockedSession = session.asMockedSession();
const join = vi.spyOn(session, "join");
const client = {
@@ -112,7 +133,8 @@ function setup(
getDeviceId: () => deviceId,
getDomain: () => "example.org",
getSyncState: () => SyncState.Syncing,
_unstable_getRTCTransports: vi.fn().mockResolvedValue([transport]),
_unstable_getRTCTransports: vi.fn().mockResolvedValue(homeserverTransports),
_unstable_getLivekitToken: vi.fn().mockResolvedValue({ jwt: publisherJWT }),
} as unknown as MatrixClient;
const members$ = new BehaviorSubject<RTCMember[]>([]);
@@ -220,6 +242,32 @@ describe("createRTCParticipation", () => {
});
});
it("publishes on another member's SFU where the homeserver advertises none", async () => {
const bobOnSfu = mockRtcMembership("@bob:example.org", "BOB", {
fociPreferred: [{ type: "livekit", url: "wss://sfu.example.org" }],
});
const { join, fake } = setup(
{},
{},
{ homeserverTransports: [], memberships: [bobOnSfu] },
);
fake.prepared.resolve({ canDelegateDelayedLeave: false });
await flushPromises();
await flushPromises();
expect(join).toHaveBeenCalledWith(
expect.anything(),
[
{
type: "livekit",
url: "wss://sfu.example.org",
server_name: "example.org",
},
],
expect.anything(),
);
});
it("publishes only while joined", async () => {
const { participation, session, fake, scope } = setup();
const ended = vi.fn();
+29 -6
View File
@@ -11,6 +11,7 @@ import {
type MatrixRTCSession as JsSdkRTCSession,
MatrixRTCSessionEvent,
MembershipManagerEvent,
type Transport,
} from "matrix-js-sdk/lib/matrixrtc";
import { type IMembershipManager } from "matrix-js-sdk/lib/matrixrtc/IMembershipManager";
import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager";
@@ -50,6 +51,7 @@ import {
type MediaKey,
} from "./media-backend/api";
import { createLivekitBackend } from "./media-backend/livekit/LivekitBackend";
import { findRemotePublishLivekitTransport } from "./media-backend/livekit/RemotePublishTransport";
import { createHomeserverConnected$ } from "./matrixrtc/HomeserverConnected";
import { joinJsSdkSession } from "./matrixrtc/joinJsSdkSession";
import {
@@ -130,13 +132,34 @@ export function createRTCParticipation(
logger,
});
// If the backend is LiveKit, we check it for transports as well, as a
// fallback: another member's SFU may let us publish (MSC4195). LiveKit only,
// so it stays out of the backend interface
const transportFromBackends = async (): Promise<Transport | null> =>
backend.transportType === "livekit"
? findRemotePublishLivekitTransport({
client,
// A snapshot: if nobody with a usable SFU is here yet, the user has
// to come back later rather than wait in a half-working lobby
memberships: jsSdkSession.memberships,
ownMembershipIdentity,
roomId: room.roomId,
slotId: jsSdkSession.slotId,
logger,
})
: null;
const preparedTransport$ = from(
discoverLocalTransport(client, backend.transportType, options, logger).then(
async (transport): Promise<PreparedTransport> => ({
transport,
...(await backend.prepareLocalTransport(transport)),
}),
),
discoverLocalTransport(
client,
backend.transportType,
options,
transportFromBackends,
logger,
).then(async (transport): Promise<PreparedTransport> => ({
transport,
...(await backend.prepareLocalTransport(transport)),
})),
);
const localMembership = createLocalMembership$({
+36 -1
View File
@@ -20,6 +20,11 @@ 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",
};
function client(
advertised: Transport[],
@@ -30,6 +35,12 @@ function client(
};
}
function fromBackends(
transport: Transport | null = null,
): ReturnType<typeof vi.fn<() => Promise<Transport | null>>> {
return vi.fn(async () => Promise.resolve(transport));
}
describe("discoverLocalTransport", () => {
it("takes the transport the host names without asking the homeserver", async () => {
const c = client([homeserverTransport]);
@@ -39,6 +50,7 @@ describe("discoverLocalTransport", () => {
c,
"livekit",
{ transport: hostTransport },
fromBackends(),
logger,
),
).resolves.toBe(hostTransport);
@@ -51,6 +63,7 @@ describe("discoverLocalTransport", () => {
client([homeserverTransport]),
"livekit",
{ fallbackTransport: hostTransport },
fromBackends(),
logger,
),
).resolves.toEqual(homeserverTransport);
@@ -62,13 +75,35 @@ describe("discoverLocalTransport", () => {
client([{ type: "not_livekit" }]),
"livekit",
{ fallbackTransport: hostTransport },
fromBackends(),
logger,
),
).resolves.toEqual(hostTransport);
});
it("prefers a local transport over the backends' without asking them", async () => {
const backends = fromBackends(bobsTransport);
await expect(
discoverLocalTransport(
client([homeserverTransport]),
"livekit",
{},
backends,
logger,
),
).resolves.toEqual(homeserverTransport);
expect(backends).not.toHaveBeenCalled();
});
it("fails naming the homeserver where there is no transport at all", async () => {
const promise = discoverLocalTransport(client([]), "livekit", {}, logger);
const promise = discoverLocalTransport(
client([]),
"livekit",
{},
fromBackends(),
logger,
);
await expect(promise).rejects.toBeInstanceOf(
MatrixRTCTransportMissingError,
+8 -2
View File
@@ -15,7 +15,10 @@ 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 the first the backends can provide, which are asked only then.
*
* @throws {MatrixRTCTransportMissingError} where none of these exists.
*/
export async function discoverLocalTransport(
client: Pick<MatrixClient, "getDomain" | "_unstable_getRTCTransports">,
@@ -24,6 +27,7 @@ export async function discoverLocalTransport(
transport: override,
fallbackTransport,
}: Pick<RTCParticipationOptions, "transport" | "fallbackTransport">,
transportFromBackends: () => Promise<Transport | null>,
logger: Logger,
): Promise<Transport> {
const transport =
@@ -33,7 +37,9 @@ export async function discoverLocalTransport(
transportType,
fallbackTransport,
logger,
}).discoverPreferredTransport());
}).discoverPreferredTransport()) ??
(await transportFromBackends()) ??
null;
if (transport === null)
throw new MatrixRTCTransportMissingError(client.getDomain() ?? "");
return transport;
@@ -258,6 +258,23 @@ 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("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 +505,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");
});
});
});
+31 -16
View File
@@ -34,7 +34,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,
@@ -146,26 +146,38 @@ export function createLivekitBackend(
logger,
);
const ownServerName = serverNameOf(ownMembershipIdentity.userId);
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 +216,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))),
@@ -0,0 +1,172 @@
/*
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 { findRemotePublishLivekitTransport } 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" };
const bobsTransport = { ...bobSfu, server_name: "bob.com" };
const carolsTransport = { ...carolSfu, server_name: "carol.com" };
// Bob is the oldest member, Carol joined after him
const bob = mockRtcMembership("@bob:bob.com", "BOB", {
fociPreferred: [bobSfu],
membership: { created_ts: 100 },
});
const carol = mockRtcMembership("@carol:carol.com", "CAROL", {
fociPreferred: [carolSfu],
membership: { created_ts: 200 },
});
// Bob's other device, and Dave who publishes on Bob's SFU remotely
const bobsLaptop = mockRtcMembership("@bob:bob.com", "LAPTOP", {
fociPreferred: [bobSfu],
membership: { created_ts: 150 },
});
const dave = mockRtcMembership("@dave:dave.com", "DAVE", {
fociPreferred: [bobsTransport],
membership: { created_ts: 160 },
});
const legacyErin = mockRtcMembership("@erin:erin.com", "ERIN", {
fociPreferred: [
{ type: "livekit", livekit_service_url: "https://jwt.erin.com" },
],
membership: { created_ts: 50 },
});
const ourselves = mockRtcMembership(
ownMemberMock.userId,
ownMemberMock.deviceId,
{
fociPreferred: [{ type: "livekit", url: "wss://sfu.example.org" }],
membership: { created_ts: 10 },
},
);
const csApiMissing = new MatrixError(
{ errcode: "M_UNRECOGNIZED", error: "" },
404,
);
describe("findRemotePublishLivekitTransport", () => {
let client: Mocked<ClientGetTokenParts>;
const find = async (
memberships: CallMembership[],
): ReturnType<typeof findRemotePublishLivekitTransport> =>
findRemotePublishLivekitTransport({
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(bobsTransport);
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("asks about an SFU once however many members are on it", async () => {
client._unstable_getLivekitToken.mockResolvedValue({
jwt: jwtWithPublishGrant(false),
});
await expect(find([dave, bobsLaptop, bob])).resolves.toBeNull();
expect(client._unstable_getLivekitToken).toHaveBeenCalledTimes(1);
});
it("skips our own membership", async () => {
await expect(find([ourselves, bob])).resolves.toEqual(bobsTransport);
expect(client._unstable_getLivekitToken).toHaveBeenCalledTimes(1);
});
it("ignores members publishing through a legacy JWT service", async () => {
await expect(find([legacyErin, bob])).resolves.toEqual(bobsTransport);
expect(client._unstable_getLivekitToken).toHaveBeenCalledTimes(1);
});
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.toEqual(carolsTransport);
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.toEqual(carolsTransport);
});
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([ourselves])).resolves.toBeNull();
expect(client._unstable_getLivekitToken).not.toHaveBeenCalled();
});
});
@@ -0,0 +1,105 @@
/*
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,
isUnstableLivekitTransport,
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 none does. Candidates are tried
* oldest membership first, each SFU once, and each is checked by asking for a
* publisher token through our homeserver (MSC4195): a homeserver may restrict
* remote publishers, which shows as a token without the publish grant.
*/
export async function findRemotePublishLivekitTransport({
client,
memberships,
ownMembershipIdentity: own,
roomId,
slotId,
logger,
}: RemotePublishTransportParams): Promise<Transport | null> {
logger.info(
"No local transport available, looking for another member's SFU to publish on",
);
const tried = new Set<string>();
const byAge = [...memberships].sort((a, b) => a.createdTs() - b.createdTs());
for (const membership of byAge) {
if (
membership.userId === own.userId &&
membership.deviceId === own.deviceId
)
continue;
for (const transport of membership.transports) {
// Publishing remotely goes through the client-server API, which a JWT
// service cannot stand in for, so only a `url` qualifies
if (!isUnstableLivekitTransport(transport) || !("url" in transport))
continue;
const { url } = transport;
const serverName = membership.getTransportServerName(transport);
const key = JSON.stringify([url, serverName]);
if (tried.has(key)) continue;
tried.add(key);
try {
const { jwt } = await csApi.getSFUConfig({
client,
membership: own,
url,
serverName,
roomId,
slotId,
});
try {
// The SFU treats an absent canPublish grant as publishing allowed.
if (decodeSFUJWTPayload(jwt).video.canPublish !== false) {
logger.info(`Publishing on remote SFU ${url} of ${serverName}`);
return { type: "livekit", url, server_name: serverName };
}
logger.info(
`${serverName} does not let us publish on ${url}, skipping`,
);
continue;
} catch (e) {
logger.info(
`${serverName}: failed to decode SFUJWT payload: ${e}, skipping`,
);
continue;
}
} catch (e) {
if (isCsApiUnavailable(e)) {
// Without the endpoint no further candidate can work
logger.warn(
`Cannot use the MSC4195 get_token endpoint (${e}), so no remote SFU can be used`,
);
return null;
}
logger.warn(`Failed to get a token for ${url} of ${serverName}:`, e);
continue;
}
}
}
return null;
}
+1 -1
View File
@@ -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);
}
+10 -3
View File
@@ -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;
}
+16
View File
@@ -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"
+1 -1
View File
@@ -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",
+2
View File
@@ -111,6 +111,8 @@ describe("widget", () => {
receiveToDevice: sendRecvToDevice,
turnServers: false,
rtcTransports: true,
rtcLivekitGetToken: true,
rtcLivekitDelegateDelayedLeave: true,
sendDelayedEvents: true,
updateDelayedEvents: true,
sendSticky: true,
+2
View File
@@ -175,6 +175,8 @@ export const initializeWidget = (
receiveToDevice: sendRecvToDevice,
turnServers: false,
rtcTransports: true,
rtcLivekitGetToken: true,
rtcLivekitDelegateDelayedLeave: true,
sendDelayedEvents: true,
updateDelayedEvents: true,
sendSticky: true,