Key connections on rooms

Make drag & drop of floating tiles less janky

docker-compose-playwright.yml

@@ -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:b2a5a30afb4f583a7560448b005d8332bee29672f15826e4a995b393058f0291
+    image: ghcr.io/element-hq/element-web:develop@sha256:79e47d8d5a93848403cb83042a405d8ed233a7727c499fa53b4b39a80f090be4
   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:b2a5a30afb4f583a7560448b005d8332bee29672f15826e4a995b393058f0291
+    image: ghcr.io/element-hq/element-web:develop@sha256:79e47d8d5a93848403cb83042a405d8ed233a7727c499fa53b4b39a80f090be4

playwright/widget/test-helpers.ts

@@ -159,7 +159,7 @@ export class TestHelpers {
   public static async dismissStartupToasts(page: Page): Promise<void> {
     const expectedToasts = [
       { title: "Failed to load service worker", button: "OK" },
-      { title: "Back up your chats", button: "Dismiss" },
+      { title: "Back up your chats", button: "Not now" },
       { title: "Turn on key storage", button: "Dismiss" },
       { title: "Element does not support this browser", button: "Dismiss" },
     ];
@@ -178,7 +178,11 @@ export class TestHelpers {
         );
 
         if (toastConfig) {
-          await toast.getByRole("button", { name: toastConfig.button }).click();
+          // Bounded so that a renamed button leaves the toast in place and the
+          // next step reports it, instead of hanging until the test timeout.
+          await toast
+            .getByRole("button", { name: toastConfig.button })
+            .click({ timeout: 5000 });
         } else {
           // Unknown toast. We don't want to act on unknown toasts
           break;

sdk/src/media-backend/livekit/Connection.test.ts

@@ -44,6 +44,7 @@ import { type ClientGetTokenParts, type ClientOpenIDParts } from "./auth";
 import { MatrixRTCError } from "../../errors";
 import { testJWTToken } from "../../utils/test-fixtures";
 import { ownMemberMock } from "../../utils/test";
+import { LivekitRooms, livekitRoomKey } from "./LivekitRooms";
 import { mockRemoteParticipant } from "./test";
 
 let testScope: ObservableScope;
@@ -74,6 +75,7 @@ function connectionOpts(): ConnectionOpts {
     serverName: "m.localhost",
     scope: testScope,
     ownMembershipIdentity: ownMemberMock,
+    rooms: new LivekitRooms(logger),
     livekitRoomFactory: () => fakeLivekitRoom,
   };
 }
@@ -711,3 +713,161 @@ describe("remote track logging", () => {
 //   });
 // });
 // });
+
+describe("LiveKit rooms", () => {
+  const roomKey = livekitRoomKey(transport.url!, "!example_room_id");
+  const otherConnection = (): Connection =>
+    ({ role: "subscriber", stop: vi.fn() }) as unknown as Connection;
+
+  it("waits for the room its token is for, and connects with a fresh token once it holds it", async () => {
+    setupTest();
+    const rooms = new LivekitRooms(logger);
+    const holder = otherConnection();
+    await rooms.claim(holder, roomKey);
+    fakeLivekitRoom.connect.mockResolvedValue(undefined);
+    const connection = new Connection({ ...connectionOpts(), rooms }, logger);
+
+    const started = connection.start();
+    await vi.waitFor(() =>
+      expect(rooms.holders$.value.get(connection)).toBe(holder),
+    );
+    expect(fakeLivekitRoom.connect).not.toHaveBeenCalled();
+    expect(client._unstable_getLivekitToken).toHaveBeenCalledTimes(1);
+
+    rooms.release(holder);
+    await started;
+    expect(client._unstable_getLivekitToken).toHaveBeenCalledTimes(2);
+    expect(fakeLivekitRoom.connect).toHaveBeenCalledWith(
+      transport.url,
+      testJWTToken,
+    );
+    expect(rooms.holders$.value.get(connection)).toBe(connection);
+  });
+
+  it("connects at once with the token it has where the room is free", async () => {
+    setupTest();
+    const rooms = new LivekitRooms(logger);
+    fakeLivekitRoom.connect.mockResolvedValue(undefined);
+    const connection = new Connection({ ...connectionOpts(), rooms }, logger);
+
+    await connection.start();
+    expect(client._unstable_getLivekitToken).toHaveBeenCalledTimes(1);
+    expect(fakeLivekitRoom.connect).toHaveBeenCalled();
+  });
+
+  it("gives the room up when stopped, without connecting", async () => {
+    setupTest();
+    const rooms = new LivekitRooms(logger);
+    const holder = otherConnection();
+    await rooms.claim(holder, roomKey);
+    fakeLivekitRoom.disconnect.mockResolvedValue(undefined);
+    const connection = new Connection({ ...connectionOpts(), rooms }, logger);
+
+    const started = connection.start();
+    await vi.waitFor(() =>
+      expect(rooms.holders$.value.get(connection)).toBe(holder),
+    );
+    await connection.stop();
+    await started;
+
+    expect(fakeLivekitRoom.connect).not.toHaveBeenCalled();
+    expect(connection.state$.value).toEqual({ state: "stopped" });
+    rooms.release(holder);
+    expect(rooms.holders$.value.get(connection)).toBeUndefined();
+  });
+
+  it("hands the room on only once it has left the SFU", async () => {
+    setupTest();
+    const rooms = new LivekitRooms(logger);
+    fakeLivekitRoom.connect.mockResolvedValue(undefined);
+    const disconnected = Promise.withResolvers<void>();
+    fakeLivekitRoom.disconnect.mockReturnValue(disconnected.promise);
+    const connection = new Connection({ ...connectionOpts(), rooms }, logger);
+    await connection.start();
+    const waiter = otherConnection();
+    const claim = rooms.claim(waiter, roomKey);
+
+    const stopping = connection.stop();
+    await Promise.resolve();
+    expect(rooms.holderOf(roomKey)).toBe(connection);
+
+    disconnected.resolve();
+    await stopping;
+    await claim;
+    expect(rooms.holderOf(roomKey)).toBe(waiter);
+  });
+
+  it("gives the room up when the connection fails", async () => {
+    setupTest();
+    const rooms = new LivekitRooms(logger);
+    fakeLivekitRoom.connect.mockRejectedValue(new Error("no"));
+    const connection = new Connection({ ...connectionOpts(), rooms }, logger);
+
+    await expect(connection.start()).rejects.toThrow();
+    expect(rooms.holders$.value.size).toBe(0);
+  });
+
+  it("connects once the connection holding its room stops", async () => {
+    const rooms = new LivekitRooms(logger);
+    setupTest();
+    const holderRoom = fakeLivekitRoom;
+    holderRoom.connect.mockResolvedValue(undefined);
+    holderRoom.disconnect.mockResolvedValue(undefined);
+    const holder = new Connection(
+      { ...connectionOpts(), rooms, livekitRoomFactory: () => holderRoom },
+      logger,
+    );
+    await holder.start();
+    setupTest();
+    const waiterRoom = fakeLivekitRoom;
+    waiterRoom.connect.mockResolvedValue(undefined);
+    const waiter = new Connection(
+      { ...connectionOpts(), rooms, livekitRoomFactory: () => waiterRoom },
+      logger,
+    );
+    const started = waiter.start();
+    await vi.waitFor(() =>
+      expect(rooms.holders$.value.get(waiter)).toBe(holder),
+    );
+    expect(waiterRoom.connect).not.toHaveBeenCalled();
+
+    await holder.stop();
+    await started;
+    expect(waiterRoom.connect).toHaveBeenCalled();
+    expect(rooms.holderOf(roomKey)).toBe(waiter);
+  });
+
+  it("lets a publisher take the room over, stopping the subscriber on it", async () => {
+    const rooms = new LivekitRooms(logger);
+    setupTest();
+    const subscriberRoom = fakeLivekitRoom;
+    subscriberRoom.connect.mockResolvedValue(undefined);
+    subscriberRoom.disconnect.mockResolvedValue(undefined);
+    const subscriber = new Connection(
+      { ...connectionOpts(), rooms, livekitRoomFactory: () => subscriberRoom },
+      logger,
+    );
+    await subscriber.start();
+    setupTest();
+    const publisherRoom = fakeLivekitRoom;
+    publisherRoom.connect.mockResolvedValue(undefined);
+    const publisher = new Connection(
+      {
+        ...connectionOpts(),
+        role: "publisher",
+        rooms,
+        livekitRoomFactory: () => publisherRoom,
+      },
+      logger,
+    );
+
+    await publisher.start();
+    expect(publisherRoom.connect).toHaveBeenCalled();
+    expect(subscriberRoom.disconnect).toHaveBeenCalled();
+    await vi.waitFor(() =>
+      expect(subscriber.state$.value).toEqual({ state: "stopped" }),
+    );
+    expect(rooms.holders$.value.get(subscriber)).toBe(publisher);
+    expect(rooms.holders$.value.get(publisher)).toBe(publisher);
+  });
+});

sdk/src/media-backend/livekit/Connection.ts

@@ -45,6 +45,7 @@ import {
   SFURoomCreationRestrictedError,
   UnknownRTCError,
 } from "../../errors";
+import { type LivekitRooms, livekitRoomKey } from "./LivekitRooms";
 
 export interface ConnectionOpts {
   /** The identity parts to use on this connection */
@@ -63,6 +64,8 @@ export interface ConnectionOpts {
   slotId: string;
   /** The observable scope to use for this connection. */
   scope: ObservableScope;
+  /** The room / connection mapping. */
+  rooms: LivekitRooms;
 
   /** Optional factory to create the LiveKit room, mainly for testing purposes. */
   livekitRoomFactory: () => LivekitRoom;
@@ -137,6 +140,7 @@ export class Connection {
   private readonly slotId: string;
   private readonly logger: Logger;
   private readonly ownMembershipIdentity: CallMembershipIdentityParts;
+  private readonly rooms: LivekitRooms;
   /**
    * Creates a new connection to a matrix RTC LiveKit backend.
    *
@@ -160,6 +164,7 @@ export class Connection {
     this.transport = transport;
     this.serverName = serverName;
     this.client = client;
+    this.rooms = opts.rooms;
 
     this.remoteParticipants$ = scope.behavior(
       // Only tracks remote participants
@@ -287,7 +292,9 @@ export class Connection {
    * This will:
    * 1. Authenticate with the transport for our role, through the homeserver
    *    or the JWT service.
-   * 2. Connect to the LiveKit room the token is for.
+   * 2. Claim the LiveKit room the token is for, waiting while another
+   *    connection holds it.
+   * 3. Connect to the room.
    *
    * The errors are also represented as a state in the `state$` observable.
    * It is safe to ignore those errors and handle them accordingly via the `state$` observable.
@@ -300,7 +307,7 @@ export class Connection {
     this.stopped = false;
     try {
       this._state$.next({ state: "authenticating" });
-      const { url, jwt, livekitAlias } = await this.getSFUConfig();
+      let { url, jwt, livekitAlias } = await this.getSFUConfig();
       this.logger.debug(
         "Starting Connection - jwt: ",
         jwt,
@@ -313,6 +320,18 @@ export class Connection {
       // If we were stopped while fetching the config, don't proceed to connect
       if (this.stopped) return;
 
+      // Try to claim the room.
+      const waited = await this.rooms.claim(
+        this,
+        livekitRoomKey(url, livekitAlias),
+      );
+      if (this.stopped) return;
+      if (waited) {
+        // Get a new token since it may have expired in the meantime.
+        ({ url, jwt, livekitAlias } = await this.getSFUConfig());
+        if (this.stopped) return;
+      }
+
       // Setup observer once we are done with getSFUConfig
       connectionStateObserver(this.livekitRoom)
         .pipe(this.scope.bind())
@@ -368,6 +387,7 @@ export class Connection {
         return;
       }
       this.logger.debug(`Failed to connect to LiveKit room: ${error}`);
+      this.rooms.release(this);
       this._state$.next({
         state: "error",
         error:
@@ -408,6 +428,8 @@ export class Connection {
     // disconnect sees the flag and does not report the abort as an error.
     this.stopped = true;
     await this.livekitRoom.disconnect();
+    // Release the room after disconnecting so that the next holder never meets us on the SFU.
+    this.rooms.release(this);
     this._state$.next({ state: "stopped" });
     this.logger.debug("stop: DONE disconnecting from lk room");
   }

sdk/src/media-backend/livekit/ConnectionFactory.test.ts

@@ -23,6 +23,7 @@ import {
   LivekitConnectionFactory,
   type ResolvedConnection,
 } from "./ConnectionFactory";
+import { LivekitRooms } from "./LivekitRooms";
 import { exampleTransport } from "./test";
 
 const urlTransport: UnstableLivekitTransport = {
@@ -56,6 +57,7 @@ function create(
     constant(audioOutputDeviceId),
     undefined,
     mediaQuality,
+    new LivekitRooms(logger),
   );
   const connection = factory.createConnection(
     testScope(),

sdk/src/media-backend/livekit/ConnectionFactory.ts

@@ -31,6 +31,7 @@ import {
 } from "./auth";
 import { Connection, type ConnectionOpts } from "./Connection";
 import { buildLiveKitOptions } from "./livekitOptions";
+import { type LivekitRooms } from "./LivekitRooms";
 import {
   audioCaptureOptions,
   type PublishRequestFor,
@@ -108,6 +109,7 @@ export class LivekitConnectionFactory implements ConnectionFactory {
     private readonly audioOutputDeviceId$: Behavior<string | undefined>,
     private readonly keyProvider: BaseKeyProvider | undefined,
     private readonly mediaQuality: MediaQuality | undefined,
+    private readonly rooms: LivekitRooms,
   ) {}
 
   public createConnection(
@@ -127,6 +129,7 @@ export class LivekitConnectionFactory implements ConnectionFactory {
         serverName,
         client: this.client,
         scope,
+        rooms: this.rooms,
         livekitRoomFactory: () =>
           new LivekitRoom(
             roomOptions(

sdk/src/media-backend/livekit/ConnectionMap.test.ts

@@ -22,6 +22,7 @@ import {
   keyFor,
 } from "./ConnectionMap.ts";
 import { type ConnectionFactory } from "./ConnectionFactory.ts";
+import { LivekitRooms, livekitRoomKey } from "./LivekitRooms";
 import { type Connection } from "./Connection.ts";
 import { ownMemberMock, withTestScheduler } from "../../utils/test";
 import { type Behavior } from "../../reactive/Behavior";
@@ -101,6 +102,7 @@ describe("connections$ stream", () => {
         remoteTransports$: behavior("a", {
           a: new Epoch([TRANSPORT_1, TRANSPORT_2], 0),
         }),
+        rooms: new LivekitRooms(logger),
         logger: logger,
         ownMembershipIdentity: ownMemberMock,
       });
@@ -138,6 +140,7 @@ describe("connections$ stream", () => {
         remoteTransports$: behavior("a", {
           a: new Epoch([TRANSPORT_2], 0),
         }),
+        rooms: new LivekitRooms(logger),
         logger: logger,
         ownMembershipIdentity: ownMemberMock,
       });
@@ -180,6 +183,7 @@ describe("connections$ stream", () => {
         remoteTransports$: behavior("a", {
           a: new Epoch([sameServiceElsewhere], 0),
         }),
+        rooms: new LivekitRooms(logger),
         logger: logger,
         ownMembershipIdentity: ownMemberMock,
       });
@@ -224,6 +228,7 @@ describe("connections$ stream", () => {
         remoteTransports$: behavior("a", {
           a: new Epoch([sameUrlElsewhere], 0),
         }),
+        rooms: new LivekitRooms(logger),
         logger: logger,
         ownMembershipIdentity: ownMemberMock,
       });
@@ -257,6 +262,7 @@ describe("connections$ stream", () => {
           e: new Epoch([TRANSPORT_1], 4),
           f: new Epoch([TRANSPORT_1, TRANSPORT_2], 5),
         }),
+        rooms: new LivekitRooms(logger),
         logger: logger,
         ownMembershipIdentity: ownMemberMock,
       });
@@ -297,6 +303,7 @@ describe("connections$ stream", () => {
           b: new Epoch([TRANSPORT_1, TRANSPORT_2], 1),
           c: new Epoch([TRANSPORT_1], 2),
         }),
+        rooms: new LivekitRooms(logger),
         logger: logger,
         ownMembershipIdentity: ownMemberMock,
       });
@@ -436,6 +443,7 @@ describe("connectionMap$", () => {
         remoteTransports$: behavior("a", {
           a: new Epoch([TRANSPORT_1, TRANSPORT_2], 0),
         }),
+        rooms: new LivekitRooms(logger),
         logger,
         ownMembershipIdentity: ownMemberMock,
       });
@@ -491,3 +499,58 @@ describe("connectionMap$", () => {
     });
   });
 });
+
+describe("one connection per LiveKit room", () => {
+  const room = livekitRoomKey("wss://sfu.example.org", "alias");
+
+  test("reads a transport through the connection holding its room", async () => {
+    const rooms = new LivekitRooms(logger);
+    const connectionMap$ = createConnectionMap$({
+      scope: testScope,
+      connectionFactory: fakeConnectionFactory,
+      localTransport: TRANSPORT_1,
+      remoteTransports$: new BehaviorSubject(new Epoch([TRANSPORT_2], 0)),
+      rooms,
+      logger,
+      ownMembershipIdentity: ownMemberMock,
+    });
+    const [conn1, conn2] = allCreatedConnections;
+    const participant = { identity: "a" } as RemoteParticipant;
+    (conn1.remoteParticipants$ as BehaviorSubject<RemoteParticipant[]>).next([
+      participant,
+    ]);
+    expect(
+      connectionMap$.value.value.getParticipantsForTransport(TRANSPORT_2),
+    ).toEqual([]);
+
+    // Both tokens name the same room: the second connection waits for it
+    await rooms.claim(conn1, room);
+    void rooms.claim(conn2, room);
+
+    const map = connectionMap$.value.value;
+    expect(map.getConnectionForTransport(TRANSPORT_2)).toBe(conn1);
+    expect(map.getParticipantsForTransport(TRANSPORT_2)).toEqual([participant]);
+    expect(map.getConnections()).toEqual([conn1, conn2]);
+  });
+
+  test("forgets a connection with its transport", async () => {
+    const rooms = new LivekitRooms(logger);
+    const remoteTransports$ = new BehaviorSubject(new Epoch([TRANSPORT_2], 0));
+    createConnectionMap$({
+      scope: testScope,
+      connectionFactory: fakeConnectionFactory,
+      localTransport: TRANSPORT_1,
+      remoteTransports$,
+      rooms,
+      logger,
+      ownMembershipIdentity: ownMemberMock,
+    });
+    const [conn1, conn2] = allCreatedConnections;
+    await rooms.claim(conn1, room);
+    void rooms.claim(conn2, room);
+    expect(rooms.holders$.value.size).toBe(2);
+
+    remoteTransports$.next(new Epoch([], 1));
+    expect(rooms.holders$.value).toEqual(new Map([[conn1, conn1]]));
+  });
+});

sdk/src/media-backend/livekit/ConnectionMap.ts

@@ -21,6 +21,7 @@ import { generateItemsWithEpoch } from "../../reactive/observable";
 import { type ConnectionRole, type TransportLocator } from "./auth";
 import { type Connection } from "./Connection";
 import { type ConnectionFactory } from "./ConnectionFactory";
+import { type LivekitRooms } from "./LivekitRooms";
 
 // Branded, so that only values returned by `keyFor` can be used as keys
 export type TransportKey = string & { __transportKey: true };
@@ -57,12 +58,17 @@ function removeDuplicateTransports(
   return [...map.values()];
 }
 
-/** The connections at one moment, with the remote participants seen on each. */
+/**
+ * The connections at one moment, with the remote participants seen on each.
+ * A transport whose connection waits for a room another connection holds
+ * reads that room through the holder.
+ */
 export class ConnectionMap {
   private readonly map = new Map<
     TransportKey,
     { connection: Connection; participants: RemoteParticipant[] }
   >();
+  private readonly holders = new Map<TransportKey, TransportKey>();
 
   public constructor(private readonly logger?: Logger) {}
 
@@ -81,20 +87,33 @@ export class ConnectionMap {
     }
   }
 
+  /** Routes a transport's lookups through the connection holding its room. */
+  public alias(connection: Connection, holder: Connection): void {
+    this.holders.set(keyFor(connection), keyFor(holder));
+  }
+
   public getConnections(): Connection[] {
     return Array.from(this.map.values()).map(({ connection }) => connection);
   }
 
+  /** The connection a transport's media arrives on or null while it has none. */
   public getConnectionForTransport(
     transport: TransportLocator,
   ): Connection | null {
-    return this.map.get(keyFor(transport))?.connection ?? null;
+    return this.entryFor(transport)?.connection ?? null;
   }
 
   public getParticipantsForTransport(
     transport: TransportLocator,
   ): RemoteParticipant[] {
-    return this.map.get(keyFor(transport))?.participants ?? [];
+    return this.entryFor(transport)?.participants ?? [];
+  }
+
+  private entryFor(
+    transport: TransportLocator,
+  ): { connection: Connection; participants: RemoteParticipant[] } | undefined {
+    const key = keyFor(transport);
+    return this.map.get(this.holders.get(key) ?? key);
   }
 }
 
@@ -105,6 +124,8 @@ interface Props {
   localTransport: TransportLocator;
   /** Every transport some member is on, deduplicated against the local one. */
   remoteTransports$: Behavior<Epoch<TransportLocator[]>>;
+  /** The room / connection mapping. */
+  rooms: LivekitRooms;
   logger: Logger;
   ownMembershipIdentity: CallMembershipIdentityParts;
 }
@@ -112,13 +133,15 @@ interface Props {
 /**
  * One connection per transport in use by one or more session members, with
  * the remote participants each one sees. The connections start as they are
- * created and stop as their transport drops out of the list.
+ * created and stop as their transport drops out of the list. Transports that
+ * resolve to one LiveKit room share the connection holding it.
  */
 export function createConnectionMap$({
   scope,
   connectionFactory,
   localTransport,
   remoteTransports$,
+  rooms,
   logger: parentLogger,
   ownMembershipIdentity,
 }: Props): Behavior<Epoch<ConnectionMap>> {
@@ -162,6 +185,7 @@ export function createConnectionMap$({
             ownMembershipIdentity,
             logger,
           );
+          scope.onEnd(() => rooms.forget(connection));
           // The state carries the progress, errors included
           void connection.start();
           return connection;
@@ -175,16 +199,23 @@ export function createConnectionMap$({
       switchMap(({ value: connections, epoch }) => {
         if (connections.length === 0)
           return of(new Epoch(new ConnectionMap(), epoch));
-        return combineLatest(
+        const participants$ = combineLatest(
           connections.map((connection) =>
             connection.remoteParticipants$.pipe(
               map((participants) => ({ connection, participants })),
             ),
           ),
-          (...list) => {
+        );
+        return combineLatest(
+          [participants$, rooms.holders$],
+          (list, holders) => {
             const map = new ConnectionMap(logger);
-            for (const item of list)
-              map.add(item.connection, item.participants);
+            for (const { connection, participants } of list) {
+              map.add(connection, participants);
+              const holder = holders.get(connection);
+              if (holder !== undefined && holder !== connection)
+                map.alias(connection, holder);
+            }
             return new Epoch(map, epoch);
           },
         );

sdk/src/media-backend/livekit/LivekitBackend.test.ts

@@ -39,6 +39,7 @@ import {
   createTransportRegistry,
   supportsDelegation,
 } from "./LivekitBackend";
+import { LivekitRooms, livekitRoomKey } from "./LivekitRooms";
 import {
   exampleServiceUrl,
   exampleTransport,
@@ -173,11 +174,14 @@ describe("createLivekitBackend", () => {
     state: "authenticated",
     livekitState: LivekitConnectionState.Connected,
   };
+  /** The registry the backend handed its factory, for the test to resolve rooms in. */
+  let rooms: LivekitRooms;
 
   /** Stands in for the factory: connections whose state and participants the test drives. */
   function stubFactory(initialState: ConnectionState): void {
     vi.spyOn(connectionFactory, "LivekitConnectionFactory").mockImplementation(
-      function (this: unknown) {
+      function (this: unknown, ...args: unknown[]) {
+        rooms = args.find((arg) => arg instanceof LivekitRooms) as LivekitRooms;
         return {
           createConnection: (
             scope: ObservableScope,
@@ -506,6 +510,64 @@ describe("createLivekitBackend", () => {
       expect(remote?.connection.serverName).toBe("sample.com");
     });
 
+    it("shares one connection between transports that resolve to the same room", async () => {
+      // Two members of one homeserver, one of them advertising the SFU's url:
+      // two transports, one JWT service, one room, and the same subscriber
+      // identity on both tokens, which the SFU would not admit twice
+      const b = await prepared();
+      const withUrl: UnstableLivekitTransport = {
+        type: "livekit",
+        url: "wss://sfu.sample.com",
+        livekit_service_url: "https://lk.sample.com",
+      };
+      const withoutUrl: UnstableLivekitTransport = {
+        type: "livekit",
+        livekit_service_url: "https://lk.sample.com",
+      };
+      const bobMedia = b.mediaFor(testScope(), {
+        ...bob(withUrl, ["@bob:sample.com:DEV1"]),
+        userId: "@bob:sample.com",
+        deviceId: "DEV1",
+      });
+      const carolMedia = b.mediaFor(testScope(), {
+        ...bob(withoutUrl, ["@carol:sample.com:DEV2"]),
+        userId: "@carol:sample.com",
+        deviceId: "DEV2",
+      });
+      await flushPromises();
+      const [, first, second] = connections;
+      expect(first.connection.transport).toEqual(withUrl);
+      expect(second.connection.transport).toEqual(withoutUrl);
+
+      const room = livekitRoomKey("wss://sfu.sample.com", "alias");
+      await rooms.claim(first.connection, room);
+      void rooms.claim(second.connection, room);
+      first.participants$.next([
+        mockRemoteParticipant({
+          identity: "@bob:sample.com:DEV1",
+          trackPublications: new Map(),
+        }),
+        mockRemoteParticipant({
+          identity: "@carol:sample.com:DEV2",
+          trackPublications: new Map(),
+        }),
+      ]);
+      await flushPromises();
+
+      expect(bobMedia.tracks$.value).toEqual([]);
+      expect(carolMedia.tracks$.value).toEqual([]);
+      // The waiting transport reports the state of the connection carrying it
+      second.state$.next({ state: "authenticating" });
+      await flushPromises();
+      expect(
+        b.connections$.value.map(({ transport, state }) => [transport, state]),
+      ).toEqual([
+        [bothTransport, MediaConnectionState.Connected],
+        [withUrl, MediaConnectionState.Connected],
+        [withoutUrl, MediaConnectionState.Connected],
+      ]);
+    });
+
     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

sdk/src/media-backend/livekit/LivekitBackend.ts

@@ -61,6 +61,7 @@ import { toMediaConnectionState } from "./connectionState";
 import { createDataChannel$ } from "./DataChannel";
 import { createKeyProvider } from "./KeyProvider";
 import { createLivekitMemberMedia } from "./LivekitMemberMedia";
+import { LivekitRooms } from "./LivekitRooms";
 import { createLivekitLocalMedia } from "./LocalMedia";
 
 export interface LivekitBackendOptions {
@@ -104,6 +105,7 @@ export function createLivekitBackend(
   const audioOutputDeviceId$ = new BehaviorSubject<string | undefined>(
     undefined,
   );
+  const rooms = new LivekitRooms(logger);
   const connectionFactory = new LivekitConnectionFactory(
     client,
     roomId,
@@ -112,6 +114,7 @@ export function createLivekitBackend(
     audioOutputDeviceId$,
     keyProvider,
     mediaQuality,
+    rooms,
   );
 
   // A new map, in a scope of its own, whenever the local transport changes:
@@ -123,6 +126,7 @@ export function createLivekitBackend(
         connectionFactory,
         localTransport,
         remoteTransports$: transports.transports$,
+        rooms,
         logger,
         ownMembershipIdentity,
       }),
@@ -259,7 +263,14 @@ export function createLivekitBackend(
       switchMap(({ value }) => {
         const connections = value.getConnections();
         if (connections.length === 0) return of([]);
-        return combineLatest(connections.map(backendConnection$));
+        return combineLatest(
+          connections.map((connection) =>
+            backendConnection$(
+              connection,
+              value.getConnectionForTransport(connection) ?? connection,
+            ),
+          ),
+        );
       }),
     ),
   );
@@ -306,14 +317,16 @@ async function authenticated(
     );
 }
 
+/** A transport's connection with the state of the connection holding its room. */
 function backendConnection$(
   connection: Connection,
+  holder: Connection,
 ): Observable<BackendConnection> {
   const resolved$: Observable<ResolvedTransport | undefined> =
     connection instanceof ResolvedConnection
       ? connection.resolved$
       : of(undefined);
-  return combineLatest([connection.state$, resolved$]).pipe(
+  return combineLatest([holder.state$, resolved$]).pipe(
     map(([state, resolved]): BackendConnection => ({
       transport: connection.transport,
       state: toMediaConnectionState(state),

sdk/src/media-backend/livekit/LivekitRooms.test.ts

@@ -0,0 +1,173 @@
+/*
+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 { logger } from "matrix-js-sdk/lib/logger";
+import { describe, expect, it, vi } from "vitest";
+
+import { type ConnectionRole } from "./auth";
+import { type Connection } from "./Connection";
+import { LivekitRooms, livekitRoomKey } from "./LivekitRooms";
+
+const room = livekitRoomKey("wss://sfu.example.org", "alias");
+const otherRoom = livekitRoomKey("wss://sfu.example.org", "other");
+
+describe("LivekitRooms", () => {
+  it("hands a free room to its first claimant at once", async () => {
+    const rooms = new LivekitRooms(logger);
+    const a = connection();
+    await expect(rooms.claim(a, room)).resolves.toBe(false);
+    expect(rooms.holders$.value).toEqual(new Map([[a, a]]));
+  });
+
+  it("makes a subscriber wait for a held room, reading it through the holder meanwhile", async () => {
+    const rooms = new LivekitRooms(logger);
+    const a = connection();
+    const b = connection();
+    await rooms.claim(a, room);
+    const held = settled(rooms.claim(b, room));
+    expect(rooms.holders$.value).toEqual(
+      new Map([
+        [a, a],
+        [b, a],
+      ]),
+    );
+    expect(await held()).toBe(false);
+
+    rooms.release(a);
+    expect(await held()).toBe(true);
+    // A released connection still reads the room through whoever holds it
+    expect(rooms.holders$.value).toEqual(
+      new Map([
+        [a, b],
+        [b, b],
+      ]),
+    );
+  });
+
+  it("lets a publisher take a room over and stops the subscriber holding it", async () => {
+    const rooms = new LivekitRooms(logger);
+    const subscriber = connection();
+    const publisher = connection("publisher");
+    await rooms.claim(subscriber, room);
+
+    await expect(rooms.claim(publisher, room)).resolves.toBe(false);
+    expect(subscriber.stop).toHaveBeenCalled();
+    // The stopped subscriber still reads the room, now through the publisher
+    rooms.release(subscriber);
+    expect(rooms.holders$.value).toEqual(
+      new Map([
+        [subscriber, publisher],
+        [publisher, publisher],
+      ]),
+    );
+  });
+
+  it("settles a waiter's claim when it is released without the room", async () => {
+    const rooms = new LivekitRooms(logger);
+    const a = connection();
+    const b = connection();
+    await rooms.claim(a, room);
+    const held = settled(rooms.claim(b, room));
+
+    rooms.release(b);
+    expect(await held()).toBe(true);
+    expect(rooms.holders$.value).toEqual(
+      new Map([
+        [a, a],
+        [b, a],
+      ]),
+    );
+  });
+
+  it("keeps rooms apart", async () => {
+    const rooms = new LivekitRooms(logger);
+    const a = connection();
+    const b = connection();
+    await rooms.claim(a, room);
+    await expect(rooms.claim(b, otherRoom)).resolves.toBe(false);
+    expect(rooms.holders$.value).toEqual(
+      new Map([
+        [a, a],
+        [b, b],
+      ]),
+    );
+  });
+
+  it("forgets a connection for good", async () => {
+    const rooms = new LivekitRooms(logger);
+    const a = connection();
+    const b = connection();
+    await rooms.claim(a, room);
+    void rooms.claim(b, room);
+
+    rooms.forget(a);
+    expect(rooms.holders$.value).toEqual(new Map([[b, b]]));
+    rooms.forget(b);
+    expect(rooms.holders$.value).toEqual(new Map());
+  });
+
+  it("ignores a release from a connection that never claimed", () => {
+    const rooms = new LivekitRooms(logger);
+    expect(() => rooms.release(connection())).not.toThrow();
+  });
+
+  it("reports whether a claimant had to wait", async () => {
+    const rooms = new LivekitRooms(logger);
+    const a = connection();
+    const b = connection();
+    await rooms.claim(a, room);
+    const claim = rooms.claim(b, room);
+    rooms.release(a);
+    await expect(claim).resolves.toBe(true);
+  });
+
+  it("names the holder of a room", async () => {
+    const rooms = new LivekitRooms(logger);
+    const a = connection();
+    expect(rooms.holderOf(room)).toBeUndefined();
+    await rooms.claim(a, room);
+    expect(rooms.holderOf(room)).toBe(a);
+  });
+});
+
+describe("livekitRoomKey", () => {
+  it("spells the SFU url one way however the token did", () => {
+    expect(livekitRoomKey("wss://sfu.example.org", "alias")).toBe(
+      livekitRoomKey("wss://sfu.example.org/", "alias"),
+    );
+    expect(livekitRoomKey("wss://sfu.example.org/", "alias")).toBe(
+      livekitRoomKey("WSS://SFU.EXAMPLE.ORG", "alias"),
+    );
+  });
+
+  it("keeps SFUs and aliases apart", () => {
+    const key = livekitRoomKey("wss://sfu.example.org", "alias");
+    expect(livekitRoomKey("wss://sfu.example.org/sfu", "alias")).not.toBe(key);
+    expect(livekitRoomKey("wss://sfu.example.org", "other")).not.toBe(key);
+  });
+
+  it("falls back to the raw string for something that is not a url", () => {
+    expect(livekitRoomKey("not a url", "alias")).toBe(
+      JSON.stringify(["not a url", "alias"]),
+    );
+  });
+});
+
+function connection(role: ConnectionRole = "subscriber"): Connection {
+  return { role, stop: vi.fn() } as unknown as Connection;
+}
+
+/** Whether a promise has settled by the time the microtasks have run. */
+function settled(promise: Promise<unknown>): () => Promise<boolean> {
+  let done = false;
+  void promise.then(() => (done = true));
+  return async () => {
+    await Promise.resolve();
+    await Promise.resolve();
+    return done;
+  };
+}

sdk/src/media-backend/livekit/LivekitRooms.ts

@@ -0,0 +1,177 @@
+/*
+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 { BehaviorSubject } from "rxjs";
+
+import { type Behavior } from "../../reactive/Behavior";
+import { type Connection } from "./Connection";
+
+// Branded, so that only values returned by `livekitRoomKey` can be used as keys.
+export type LivekitRoomKey = string & { __livekitRoomKey: true };
+
+/** A LiveKit room identified by the SFU it is on and its alias there. */
+export function livekitRoomKey(url: string, alias: string): LivekitRoomKey {
+  return JSON.stringify([normalisedUrl(url), alias]) as LivekitRoomKey;
+}
+
+/** Normalises URLs to make them comparable. */
+function normalisedUrl(url: string): string {
+  try {
+    return new URL(url).href;
+  } catch {
+    return url;
+  }
+}
+
+/** A room some connection has resolved to, and who is on it. */
+interface Room {
+  /** The one connection joined to the room on the SFU. */
+  holder: Connection;
+  /** Connections waiting to hold the room, in the order they will get it. */
+  waiting: Waiter[];
+}
+
+/** A connection's pending claim on a room. */
+interface Waiter {
+  /**
+   * The connection that resolved to the room and is not connecting until it
+   * holds it.
+   */
+  connection: Connection;
+  /**
+   * Settles the connection's claim after it has either acquired hold of the
+   * room or was released.
+   */
+  held: () => void;
+}
+
+/** Helper to organise rooms and holding or waiting connections. */
+export class LivekitRooms {
+  private readonly logger: Logger;
+  /** The rooms some connection is on or waiting for. */
+  private readonly rooms = new Map<LivekitRoomKey, Room>();
+  /** The room each connection resolved to. */
+  private readonly roomOf = new Map<Connection, LivekitRoomKey>();
+  /** The source of `holders$`. */
+  private readonly holders = new BehaviorSubject<Map<Connection, Connection>>(
+    new Map(),
+  );
+  /** Every connection that resolved a room, to the connection holding that room. */
+  public readonly holders$: Behavior<Map<Connection, Connection>> =
+    this.holders;
+
+  public constructor(parentLogger: Logger) {
+    this.logger = parentLogger.getChild("[LivekitRooms]");
+  }
+
+  /** The connection holding a room, if any does. */
+  public holderOf(key: LivekitRoomKey): Connection | undefined {
+    return this.rooms.get(key)?.holder;
+  }
+
+  /**
+   * Resolves once `connection` either holds the room or was released.
+   * Resolves to whether the connection had to wait for the room.
+   */
+  public async claim(
+    connection: Connection,
+    key: LivekitRoomKey,
+  ): Promise<boolean> {
+    this.roomOf.set(connection, key);
+    const room = this.rooms.get(key);
+    if (room === undefined) {
+      this.logger.debug(`${describe(connection)} holds room ${key}`);
+      this.rooms.set(key, { holder: connection, waiting: [] });
+      this.publish();
+      return false;
+    }
+    // If the connection is for publishing and a subscriber holds the room, demote it.
+    if (connection.role === "publisher") {
+      const demoted = room.holder;
+      this.logger.info(
+        `${describe(connection)} takes room ${key} over from ${describe(demoted)}, which stops`,
+      );
+      room.holder = connection;
+      this.publish();
+      void demoted.stop();
+      return false;
+    }
+    this.logger.info(
+      `${describe(connection)} waits for room ${key}, held by ${describe(room.holder)}`,
+    );
+    await new Promise<void>((held) => {
+      room.waiting.push({ connection, held });
+      this.publish();
+    });
+    return true;
+  }
+
+  /** Gives up holding or waiting for the room. */
+  public release(connection: Connection): void {
+    const key = this.roomOf.get(connection);
+    const room = key === undefined ? undefined : this.rooms.get(key);
+    if (key === undefined || room === undefined) {
+      return; // Never claimed, or the room has already emptied out
+    }
+    if (room.holder === connection) {
+      // We're the room's holder. Figure out the first waiter and promote it.
+      const next = room.waiting.shift();
+      if (next === undefined) {
+        this.rooms.delete(key);
+      } else {
+        this.logger.info(
+          `${describe(next.connection)} holds room ${key} now, released by ${describe(connection)}`,
+        );
+        room.holder = next.connection;
+        next.held();
+      }
+    } else {
+      // We're a waiter. Figure out our index in the waiting list to remove ourselves.
+      const index = room.waiting.findIndex((w) => w.connection === connection);
+      if (index === -1) {
+        return;
+      }
+      const [{ held }] = room.waiting.splice(index, 1);
+      held();
+    }
+    this.publish();
+  }
+
+  /** Releases a connection that is gone for good, so nothing reads through it. */
+  public forget(connection: Connection): void {
+    this.release(connection);
+    this.roomOf.delete(connection);
+    this.publish();
+  }
+
+  /** Emits `holders$` afresh if needed. */
+  private publish(): void {
+    const holders = new Map<Connection, Connection>();
+    for (const [connection, key] of this.roomOf) {
+      const holder = this.rooms.get(key)?.holder;
+      if (holder !== undefined) holders.set(connection, holder);
+    }
+    if (!sameHolders(this.holders.value, holders)) this.holders.next(holders);
+  }
+}
+
+/** Checks if two holder maps are identical. */
+function sameHolders(
+  a: Map<Connection, Connection>,
+  b: Map<Connection, Connection>,
+): boolean {
+  return (
+    a.size === b.size &&
+    [...a].every(([connection, holder]) => b.get(connection) === holder)
+  );
+}
+
+/** Summarises a connection for logging. */
+function describe(connection: Connection): string {
+  return `${connection.role} connection to ${JSON.stringify(connection.transport)}`;
+}

src/AppBar.module.css

@@ -7,6 +7,7 @@
   padding-top: env(safe-area-inset-top);
   opacity: 1;
   transition: opacity 0.15s;
+  text-align: center;
 }
 
 /* Pseudo-element for the gradient background */
@@ -52,7 +53,7 @@
     ". . ."
     "primaryButton title secondaryButton"
     ". . .";
-  place-items: center;
+  place-items: center stretch;
   column-gap: var(--cpd-space-2x);
 }