diff --git a/src/state/CallViewModel/remoteMembers/ConnectionMap.test.ts b/src/state/CallViewModel/remoteMembers/ConnectionMap.test.ts index 90c0e4c67..2344c1e55 100644 --- a/src/state/CallViewModel/remoteMembers/ConnectionMap.test.ts +++ b/src/state/CallViewModel/remoteMembers/ConnectionMap.test.ts @@ -127,6 +127,82 @@ describe("connections$ stream", () => { }); }); + test("Shares one connection between homeservers advertising the same JWT service", () => { + withTestScheduler(({ behavior, expectObservable }) => { + const sameServiceElsewhere: TransportLocator = { + transport: TRANSPORT_1.transport, + serverName: "elsewhere.org", + }; + const connectionMap$ = createConnectionMap$({ + scope: testScope, + connectionFactory: fakeConnectionFactory, + localTransport: TRANSPORT_1, + remoteTransports$: behavior("a", { + a: new Epoch([sameServiceElsewhere], 0), + }), + logger: logger, + ownMembershipIdentity: ownMemberMock, + }); + + expectObservable( + connectionMap$.pipe(mapEpoch((d) => d.getConnections())), + ).toBe("a", { + a: expect.toSatisfy((e: Epoch) => { + expect(e.value.length).toBe(1); + // The JWT service is reached directly, so the server name plays no + // part and the local locator, which came first, is the one used + expect( + vi.mocked(fakeConnectionFactory).createConnection, + ).toHaveBeenCalledExactlyOnceWith( + expect.anything(), + "publisher", + TRANSPORT_1.transport, + "example.org", + expect.anything(), + expect.anything(), + ); + return true; + }), + }); + }); + }); + + test("Keeps one connection per homeserver for transports with a url", () => { + withTestScheduler(({ behavior, expectObservable }) => { + const urlTransport: TransportLocator = { + transport: { type: "livekit", url: "wss://sfu.example.org" }, + serverName: "example.org", + }; + const sameUrlElsewhere: TransportLocator = { + transport: urlTransport.transport, + serverName: "elsewhere.org", + }; + const connectionMap$ = createConnectionMap$({ + scope: testScope, + connectionFactory: fakeConnectionFactory, + localTransport: urlTransport, + remoteTransports$: behavior("a", { + a: new Epoch([sameUrlElsewhere], 0), + }), + logger: logger, + ownMembershipIdentity: ownMemberMock, + }); + + expectObservable( + connectionMap$.pipe(mapEpoch((d) => d.getConnections())), + ).toBe("a", { + a: expect.toSatisfy((e: Epoch) => { + // Our homeserver asks the named server for the token, so each + // server name needs its own connection + expect(e.value.length).toBe(2); + expect(findConnection(urlTransport, e.value)).toBeDefined(); + expect(findConnection(sameUrlElsewhere, e.value)).toBeDefined(); + return true; + }), + }); + }); + }); + test("Should start connection only once", () => { withTestScheduler(({ behavior, expectObservable }) => { const connectionMap$ = createConnectionMap$({ @@ -228,7 +304,7 @@ describe("ConnectionMap", () => { data.add({ ...connection } as Connection, [p("b")]); expect(warn).toHaveBeenCalledWith( expect.stringContaining( - 'second connection to [null,"https://lk.example.org","example.org"]: existing [a], adding [b]', + 'second connection to [null,"https://lk.example.org",null]: existing [a], adding [b]', ), ); expect(data.getParticipantsForTransport(TRANSPORT_1)).toHaveLength(2); diff --git a/src/state/CallViewModel/remoteMembers/ConnectionMap.ts b/src/state/CallViewModel/remoteMembers/ConnectionMap.ts index 4adc92b04..586e8d49b 100644 --- a/src/state/CallViewModel/remoteMembers/ConnectionMap.ts +++ b/src/state/CallViewModel/remoteMembers/ConnectionMap.ts @@ -6,7 +6,6 @@ SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial Please see LICENSE in the repository root for full details. */ -import { type UnstableLivekitTransport } from "matrix-js-sdk/lib/matrixrtc"; import { combineLatest, map, of, switchMap, tap } from "rxjs"; import { type Logger } from "matrix-js-sdk/lib/logger"; import { type RemoteParticipant } from "livekit-client"; @@ -23,46 +22,40 @@ import { generateItemsWithEpoch } from "../../../utils/observable.ts"; import { type ConnectionFactory } from "./ConnectionFactory.ts"; import { type TransportLocator } from "../../../livekit/auth/types.ts"; -// Branded string type, to enforce that `fromKey` must only be called with -// values returned by `keyFor` +// Branded string type, so that only values returned by `keyFor` can be used +// as keys export type TransportKey = string & { __transportKey: true }; /** * Converts a {@link TransportLocator} to a key suitable for identifying it in a - * {@link Map} or {@link Set}. + * {@link Map} or {@link Set}. Locators with the same key share a connection. */ export function keyFor({ transport, serverName, }: TransportLocator): TransportKey { + // The server name only takes part in authentication for a transport with a + // `url`, where our homeserver asks that server for the token. A JWT service + // is reached directly, so members of several homeservers advertising the + // same one must share a connection rather than each get their own. + const hasUrl = "url" in transport; // Serialize the fields in a consistent order return JSON.stringify([ - "url" in transport ? transport.url : null, + hasUrl ? transport.url : null, "livekit_service_url" in transport ? transport.livekit_service_url : null, - serverName, + hasUrl ? serverName : null, ]) as TransportKey; } -/** - * Inverse of {@link keyFor}: turns a key back into a {@link TransportLocator}. - */ -export function fromKey(key: TransportKey): TransportLocator { - const [url, serviceUrl, serverName] = JSON.parse(key); - return { - transport: { - type: "livekit", - ...(url === null ? {} : { url }), - ...(serviceUrl === null ? {} : { livekit_service_url: serviceUrl }), - } as UnstableLivekitTransport, - serverName, - }; -} - function removeDuplicateTransports( transports: TransportLocator[], ): TransportLocator[] { + // The first locator wins, so the local one stays first in the list const map = new Map(); - for (const transport of transports) map.set(keyFor(transport), transport); + for (const transport of transports) { + const key = keyFor(transport); + if (!map.has(key)) map.set(key, transport); + } return [...map.values()]; } @@ -173,15 +166,14 @@ export function createConnectionMap$({ ? ("publisher" as const) : ("subscriber" as const); yield { - // In order to maintain a static set of map keys, we must - // temporarily serialize `transport` into a stable key format + // The locator is an object, so a stable key stands in for it keys: [role, keyFor(transport)], - data: undefined, + data: transport, }; } }, - (scope, _data$, role, transportKey) => { - const { transport, serverName } = fromKey(transportKey); + (scope, locator$, role, _key) => { + const { transport, serverName } = locator$.value; const connection = connectionFactory.createConnection( scope, role,