Mark the "all possible transport shapes at once" test as slow

Rather than whichever participant has won the race.

playwright/url-transport.spec.ts

@@ -165,6 +165,7 @@ test("Group call with all possible transport shapes at once", async ({
   page,
   browserName,
 }) => {
+  test.slow();
   skipOnFirefox(browserName);
 
   // The host starts the call with a url transport.

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

@@ -60,7 +60,10 @@ import {
 import { toMediaConnectionState } from "./connectionState";
 import { createDataChannel$ } from "./DataChannel";
 import { createKeyProvider } from "./KeyProvider";
-import { createLivekitMemberMedia } from "./LivekitMemberMedia";
+import {
+  createLivekitMemberMedia,
+  participantSourcesEqual,
+} from "./LivekitMemberMedia";
 import { createLivekitLocalMedia } from "./LocalMedia";
 
 export interface LivekitBackendOptions {
@@ -233,22 +236,28 @@ export function createLivekitBackend(
           const connection = connections.getConnectionForTransport(locator);
           // A member advertising both a `url` and a `livekit_service_url` may
           // have authenticated through either flow, under either identity
-          const participant = connections
+          const participants = connections
             .getParticipantsForTransport(locator)
-            .find((p) => backendIdentities.includes(p.identity));
-          return connection && participant
-            ? { participant, room: connection.livekitRoom }
+            .filter((p) => backendIdentities.includes(p.identity));
+          return connection && participants.length
+            ? { participants, room: connection.livekitRoom }
             : null;
         }),
-        distinctUntilChanged(
-          (a, b) => a?.participant === b?.participant && a?.room === b?.room,
-        ),
+        distinctUntilChanged(participantSourcesEqual),
       ),
     );
     source$.pipe(memberScope.bind()).subscribe((source) => {
-      logger.info(
-        `${member.rtcBackendIdentity}: LiveKit participant ${source ? `matched (${source.participant.sid})` : "missing"} on ${locator$.value ? JSON.stringify(locator$.value.transport) : "no transport"}`,
-      );
+      const transportString = locator$.value
+        ? JSON.stringify(locator$.value.transport)
+        : "no transport";
+      if (source)
+        logger.info(
+          `${member.rtcBackendIdentity}: ${source.participants.length} LiveKit participant(s) matched (${source.participants.map((p) => p.sid).join(", ")}) on ${transportString}`,
+        );
+      else
+        logger.info(
+          `${member.rtcBackendIdentity}: LiveKit participant missing on ${transportString}`,
+        );
     });
     return createLivekitMemberMedia(memberScope, source$, encryptionSystem);
   };

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

@@ -25,7 +25,7 @@ describe("createLivekitMemberMedia", () => {
     const { participant, publications, room, emitter } = fakes();
     const media = createLivekitMemberMedia(
       testScope(),
-      constant({ participant, room }),
+      constant({ participants: [participant], room }),
       { kind: E2eeType.NONE },
     );
     expect(media.tracks$.value).toEqual([]);
@@ -52,7 +52,7 @@ describe("createLivekitMemberMedia", () => {
     const { participant, publications, room, emitter } = fakes();
     const media = createLivekitMemberMedia(
       testScope(),
-      constant({ participant, room }),
+      constant({ participants: [participant], room }),
       { kind: E2eeType.NONE },
     );
     emitter.emit(

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

@@ -16,13 +16,20 @@ import {
   RoomEvent,
   type TrackPublication,
 } from "livekit-client";
-import { EMPTY, filter, map, type Observable, of, switchMap } from "rxjs";
+import {
+  EMPTY,
+  filter,
+  map,
+  type Observable,
+  of,
+  switchMap,
+  combineLatest,
+} from "rxjs";
 
 import { type ObservableScope } from "../../reactive/ObservableScope";
 import { generateItems } from "../../reactive/observable";
 import { E2eeType, type EncryptionSystem } from "../../encryption";
 import { type Behavior } from "../../reactive/Behavior";
-import { mapScoped } from "../../utils/mapScoped";
 import {
   type EncryptionError,
   type LocalMemberMedia,
@@ -33,13 +40,27 @@ import {
   createLivekitMediaTrack,
   createLocalLivekitMediaTrack,
 } from "./LivekitMediaTrack";
+import { shallowEquals as arrayShallowEquals } from "../../utils/array";
 
 /** Where a member's media comes from: its participant on a room, null while it has none. */
 export type ParticipantSource<P extends Participant = Participant> = {
-  participant: P;
+  participants: P[];
   room: LivekitRoom;
 } | null;
 
+export function participantSourcesEqual<P extends Participant = Participant>(
+  a: ParticipantSource<P>,
+  b: ParticipantSource<P>,
+): boolean {
+  return (
+    a === b ||
+    (a !== null &&
+      b !== null &&
+      arrayShallowEquals(a.participants, b.participants) &&
+      a.room === b.room)
+  );
+}
+
 /**
  * A member's media from its participant. The only place, with
  * `LivekitMediaTrack`, that reads a LiveKit participant; everything above
@@ -99,21 +120,28 @@ function memberTracks$<P extends Participant, T>(
   ) => T,
 ): Behavior<T[] | null> {
   return scope.behavior(
-    mapScoped(scope, source$, (participantScope, { participant, room }) =>
-      participantScope.behavior(
-        observeParticipantMedia(participant).pipe(
-          generateItems(
-            `${participant.identity} tracks$`,
-            function* () {
-              for (const publication of participant.trackPublications.values())
-                yield { keys: [publication.trackSid], data: publication };
-            },
-            (trackScope, publication$) =>
-              factory(trackScope, participant, publication$.value, room),
+    source$.pipe(
+      switchMap((source) => {
+        if (source === null) return of(null);
+        const { participants, room } = source;
+        return combineLatest(
+          participants.map((participant) =>
+            observeParticipantMedia(participant).pipe(
+              generateItems(
+                `${participant.identity} tracks$`,
+                function* () {
+                  for (const publication of participant.trackPublications.values())
+                    yield { keys: [publication.trackSid], data: publication };
+                },
+                (trackScope, publication$) =>
+                  factory(trackScope, participant, publication$.value, room),
+              ),
+            ),
           ),
-        ),
-      ),
-    ).pipe(switchMap((tracks$) => tracks$ ?? of(null))),
+          (...tracks) => tracks.flat(1),
+        );
+      }),
+    ),
   );
 }
 
@@ -133,7 +161,7 @@ function encryptionErrors$(
             filter(
               (message) =>
                 encryptionSystem.kind === E2eeType.SHARED_KEY ||
-                message.includes(source.participant.identity),
+                source.participants.some((p) => message.includes(p.identity)),
             ),
             map((message): EncryptionError | undefined =>
               message.includes("MissingKey")

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

@@ -10,7 +10,6 @@ import {
   type LocalParticipant,
   type LocalTrackPublication,
   ParticipantEvent,
-  type Room as LivekitRoom,
 } from "livekit-client";
 import { type Logger } from "matrix-js-sdk/lib/logger";
 import {
@@ -43,7 +42,11 @@ import { type LocalMediaBackend, MediaConnectionState } from "../api";
 import { type TransportLocator } from "./auth";
 import { type Connection } from "./Connection";
 import { type ConnectionMap } from "./ConnectionMap";
-import { createLocalLivekitMemberMedia } from "./LivekitMemberMedia";
+import {
+  createLocalLivekitMemberMedia,
+  type ParticipantSource,
+  participantSourcesEqual,
+} from "./LivekitMemberMedia";
 import { livekitSources } from "./LivekitMediaTrack";
 import { type DesiredMedia, Publisher } from "./Publisher";
 import { toMediaConnectionState } from "./connectionState";
@@ -217,17 +220,15 @@ function participantAndRoom$(
   scope: ObservableScope,
   participant$: Behavior<LocalParticipant | null>,
   connection$: Behavior<Connection | null>,
-): Behavior<{ participant: LocalParticipant; room: LivekitRoom } | null> {
+): Behavior<ParticipantSource<LocalParticipant>> {
   return scope.behavior(
     combineLatest([participant$, connection$]).pipe(
       map(([participant, connection]) =>
         participant && connection
-          ? { participant, room: connection.livekitRoom }
+          ? { participants: [participant], room: connection.livekitRoom }
           : null,
       ),
-      distinctUntilChanged(
-        (a, b) => a?.participant === b?.participant && a?.room === b?.room,
-      ),
+      distinctUntilChanged(participantSourcesEqual),
     ),
   );
 }

sdk/src/utils/array.ts

@@ -0,0 +1,16 @@
+/*
+Copyright 2024 New Vector Ltd.
+
+SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
+Please see LICENSE in the repository root for full details.
+*/
+
+/**
+ * Determine whether two arrays are equal by shallow comparison.
+ */
+export function shallowEquals<A>(first: A[], second: A[]): boolean {
+  if (first.length !== second.length) return false;
+  for (let i = 0; i < first.length; i++)
+    if (first[i] !== second[i]) return false;
+  return true;
+}