diff --git a/src/state/CallViewModel/remoteMembers/ConnectionManager.test.ts b/src/state/CallViewModel/remoteMembers/ConnectionManager.test.ts index fada34be2..809c8bccb 100644 --- a/src/state/CallViewModel/remoteMembers/ConnectionManager.test.ts +++ b/src/state/CallViewModel/remoteMembers/ConnectionManager.test.ts @@ -14,7 +14,7 @@ import { logger } from "matrix-js-sdk/lib/logger"; import { Epoch, mapEpoch, ObservableScope } from "../../ObservableScope.ts"; import { createConnectionManager$, - type ConnectionManagerData, + ConnectionManagerData, } from "./ConnectionManager.ts"; import { type ConnectionFactory } from "./ConnectionFactory.ts"; import { type Connection } from "./Connection.ts"; @@ -203,6 +203,24 @@ describe("connections$ stream", () => { }); }); +describe("ConnectionManagerData", () => { + test("warns when a second connection to the same URL is merged", () => { + const warn = vi.spyOn(logger, "warn").mockImplementation(() => {}); + const data = new ConnectionManagerData(logger); + const connection = { transport: TRANSPORT_1 } as unknown as Connection; + const p = (identity: string): RemoteParticipant => + ({ identity }) as unknown as RemoteParticipant; + data.add(connection, [p("a")]); + data.add({ ...connection } as Connection, [p("b")]); + expect(warn).toHaveBeenCalledWith( + expect.stringContaining( + "second connection to https://lk.example.org: existing [a], adding [b]", + ), + ); + expect(data.getParticipantsForTransport(TRANSPORT_1)).toHaveLength(2); + }); +}); + describe("connectionManagerData$ stream", () => { // Used in test to control fake connections' remoteParticipants$ streams let fakeRemoteParticipantsStreams: Map>; diff --git a/src/state/CallViewModel/remoteMembers/ConnectionManager.ts b/src/state/CallViewModel/remoteMembers/ConnectionManager.ts index 727f68bcc..5ebd2ce6d 100644 --- a/src/state/CallViewModel/remoteMembers/ConnectionManager.ts +++ b/src/state/CallViewModel/remoteMembers/ConnectionManager.ts @@ -30,7 +30,7 @@ export class ConnectionManagerData { { connection: Connection; participants: RemoteParticipant[] } > = new Map(); - public constructor() {} + public constructor(private readonly logger?: Logger) {} public add(connection: Connection, participants: RemoteParticipant[]): void { const key = this.getKey(connection.transport); @@ -38,6 +38,11 @@ export class ConnectionManagerData { if (!existing) { this.store.set(key, { connection, participants }); } else { + // Transports are deduplicated by URL upstream, so this should never + // happen; if it does, members may be matched against the wrong room. + this.logger?.warn( + `Merging participants from a second connection to ${key}: existing [${existing.participants.map((p) => p.identity).join(", ")}], adding [${participants.map((p) => p.identity).join(", ")}]`, + ); existing.participants.push(...participants); } } @@ -239,7 +244,7 @@ export function createConnectionManager$({ lists.reduce((data, { connection, participants }) => { data.add(connection, participants); return data; - }, new ConnectionManagerData()), + }, new ConnectionManagerData(logger)), epoch, ), ), diff --git a/src/state/CallViewModel/remoteMembers/MatrixLivekitMembers.test.ts b/src/state/CallViewModel/remoteMembers/MatrixLivekitMembers.test.ts index 09c827896..fe7621121 100644 --- a/src/state/CallViewModel/remoteMembers/MatrixLivekitMembers.test.ts +++ b/src/state/CallViewModel/remoteMembers/MatrixLivekitMembers.test.ts @@ -111,7 +111,7 @@ test("should signal participant not yet connected to livekit", async () => { }, ); expect(info).toHaveBeenCalledWith( - "[RemoteMatrixLivekitMembers] @bob:example.org:DEV000: LiveKit participant missing", + "[RemoteMatrixLivekitMembers] @bob:example.org:DEV000: LiveKit participant missing on no connection", ); }); diff --git a/src/state/CallViewModel/remoteMembers/MatrixLivekitMembers.ts b/src/state/CallViewModel/remoteMembers/MatrixLivekitMembers.ts index e884d3821..76f4670ec 100644 --- a/src/state/CallViewModel/remoteMembers/MatrixLivekitMembers.ts +++ b/src/state/CallViewModel/remoteMembers/MatrixLivekitMembers.ts @@ -111,13 +111,17 @@ export function createRemoteMatrixLivekitMembers$({ const participants = transport ? managerData.getParticipantsForTransport(transport) : []; - const participant = - participants.find( - (p) => p.identity == membership.rtcBackendIdentity, - ) ?? null; + const matches = participants.filter( + (p) => p.identity == membership.rtcBackendIdentity, + ); + const participant = matches[0] ?? null; const connection = transport ? managerData.getConnectionForTransport(transport) : null; + if (matches.length > 1) + logger.warn( + `[RemoteMatrixLivekitMembers] ${membership.rtcBackendIdentity}: ${matches.length} LiveKit participants match (sids ${matches.map((p) => p.sid).join(", ")}), using ${participant?.sid}`, + ); yield { // This could just be the backend identity without the other keys. @@ -139,8 +143,9 @@ export function createRemoteMatrixLivekitMembers$({ // Log whether the member could be matched to a LiveKit participant, // since a tile shows "waiting for media" for as long as it cannot. participant$.pipe(scope.bind()).subscribe((p) => { + const url = data$.value.connection?.transport.livekit_service_url; logger.info( - `[RemoteMatrixLivekitMembers] ${rtcBackendIdentity}: LiveKit participant ${p ? `matched (${p.sid})` : "missing"}`, + `[RemoteMatrixLivekitMembers] ${rtcBackendIdentity}: LiveKit participant ${p ? `matched (${p.sid})` : "missing"} on ${url ?? "no connection"}`, ); }); // will only get called once per backend identity. diff --git a/src/state/ObservableScope.test.ts b/src/state/ObservableScope.test.ts index 24b5d268d..12bd39ced 100644 --- a/src/state/ObservableScope.test.ts +++ b/src/state/ObservableScope.test.ts @@ -269,4 +269,26 @@ describe("behavior", () => { expect(seen.at(-1)).toBe(1); scope.end(); }); + + it("warns when the scope is ended while a behavior is mid-delivery", () => { + const warn = vi.spyOn(logger, "warn").mockImplementation(() => {}); + const scope = new ObservableScope(); + const source$ = new Subject(); + const behavior$ = scope.behavior(source$, 0); + behavior$.subscribe((v) => { + if (v === 1) scope.end(); + }); + // A derived behavior is bound to the scope, so ending the scope + // mid-delivery unsubscribes it before the value reaches it. + const derived$ = scope.behavior(behavior$); + + source$.next(1); + + expect(warn).toHaveBeenCalledWith( + expect.stringContaining("Scope ended while 1 of its behaviors"), + expect.any(String), + ); + expect(behavior$.value).toBe(1); + expect(derived$.value).toBe(0); + }); }); diff --git a/src/state/ObservableScope.ts b/src/state/ObservableScope.ts index d0cfb7e93..8f09cb6ef 100644 --- a/src/state/ObservableScope.ts +++ b/src/state/ObservableScope.ts @@ -36,6 +36,12 @@ const nothing = Symbol("nothing"); * A scope which limits the execution lifetime of its bound Observables. */ export class ObservableScope { + /** + * Number of this scope's behaviors currently mid-delivery. Used to detect a + * scope being torn down synchronously from within one of its own emissions. + */ + private delivering = 0; + private readonly ended$ = new BehaviorSubject(false); private readonly bindImpl: MonoTypeOperator = takeUntil( @@ -89,7 +95,7 @@ export class ObservableScope { let depth = 0; let maxReportedDepth = 1; setValue$.pipe(this.bind(), distinctUntilChanged()).subscribe({ - next(value) { + next: (value) => { if (depth > 0 && depth + 1 > maxReportedDepth) { maxReportedDepth = depth + 1; logger.warn( @@ -100,13 +106,15 @@ export class ObservableScope { ); } depth++; + this.delivering++; try { subject$.next(value); } finally { depth--; + this.delivering--; } }, - error(err: unknown) { + error: (err: unknown) => { subject$.error(err); }, }); @@ -119,6 +127,11 @@ export class ObservableScope { * Ends the scope, causing any bound Observables to complete. */ public end(): void { + if (this.delivering > 0) + logger.warn( + `Scope ended while ${this.delivering} of its behaviors were still delivering a value; later subscribers will be left with a stale value`, + new Error().stack, + ); this.ended$.next(true); }