From f8f5caaa0990096b45ae39343ec9b525808ac7fe Mon Sep 17 00:00:00 2001 From: Matthew Hodgson Date: Thu, 3 Sep 2026 16:50:30 +0100 Subject: [PATCH 1/6] Log remote participant and track events per LiveKit connection Rageshakes contained nothing about the state of remote tracks, so a tile showing the wrong mute or video state for a member could not be diagnosed. Log connect/disconnect, publish/unpublish, subscribe/ unsubscribe, subscription failures, remote mute/unmute and stream state changes on each Connection's logger, and remove the listeners when the connection scope ends. --- .../remoteMembers/Connection.test.ts | 78 +++++++++++++++- .../CallViewModel/remoteMembers/Connection.ts | 88 +++++++++++++++++++ 2 files changed, 165 insertions(+), 1 deletion(-) diff --git a/src/state/CallViewModel/remoteMembers/Connection.test.ts b/src/state/CallViewModel/remoteMembers/Connection.test.ts index 723295853..9d098d445 100644 --- a/src/state/CallViewModel/remoteMembers/Connection.test.ts +++ b/src/state/CallViewModel/remoteMembers/Connection.test.ts @@ -19,14 +19,17 @@ import { import { type LocalParticipant, type RemoteParticipant, + type RemoteTrack, + type RemoteTrackPublication, type Room as LivekitRoom, RoomEvent, + Track, ConnectionState as LivekitConnectionState, } from "livekit-client"; import fetchMock from "fetch-mock"; import EventEmitter from "events"; import { type IOpenIDToken } from "matrix-js-sdk"; -import { logger } from "matrix-js-sdk/lib/logger"; +import { logger, type Logger } from "matrix-js-sdk/lib/logger"; import { type LivekitTransportConfig } from "matrix-js-sdk/lib/matrixrtc"; import { @@ -509,6 +512,79 @@ describe("remote participants", () => { }); }); +describe("remote track logging", () => { + it("logs remote participant and track events on the connection logger", () => { + setupTest(); + const info = vi.fn(); + const testLogger = { + getChild: (): unknown => testLogger, + info, + debug: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + } as unknown as Logger; + new Connection( + { + client, + roomId: ROOM_ID, + transport: livekitFocus, + scope: testScope, + ownMembershipIdentity: ownMemberMock, + livekitRoomFactory: () => fakeLivekitRoom, + }, + testLogger, + ); + info.mockClear(); // drop the constructor log line + + const bob = mockRemoteParticipant({ + identity: "@bob:example.org:DEV111", + sid: "PA_bob", + }); + const pub = { + kind: "audio", + source: "microphone", + trackSid: "TR_mic", + isMuted: false, + } as unknown as RemoteTrackPublication; + const messages = (): string[] => info.mock.calls.map((c) => c[0] as string); + + fakeLivekitRoom.emit(RoomEvent.ParticipantConnected, bob); + fakeLivekitRoom.emit( + RoomEvent.TrackSubscribed, + {} as RemoteTrack, + pub, + bob, + ); + fakeLivekitRoom.emit(RoomEvent.TrackMuted, pub, bob); + fakeLivekitRoom.emit( + RoomEvent.TrackStreamStateChanged, + pub, + Track.StreamState.Paused, + bob, + ); + expect(messages()).toEqual([ + "Participant connected: @bob:example.org:DEV111 (PA_bob)", + "Subscribed: audio microphone TR_mic of @bob:example.org:DEV111 muted=false", + "Muted: audio microphone TR_mic of @bob:example.org:DEV111", + "Stream paused: audio microphone TR_mic of @bob:example.org:DEV111", + ]); + + // Local mute events are already logged by the Publisher + fakeLivekitRoom.emit(RoomEvent.TrackMuted, pub, { + ...fakeLocalParticipant, + isLocal: true, + } as unknown as LocalParticipant); + expect(messages().filter((m) => m.startsWith("Muted"))).toHaveLength(1); + + // Listeners are removed when the scope ends + testScope.end(); + fakeLivekitRoom.emit(RoomEvent.ParticipantDisconnected, bob); + expect( + messages().filter((m) => m.startsWith("Participant disconnected")), + ).toHaveLength(0); + }); +}); + // // NOT USED ANYMORE ? // diff --git a/src/state/CallViewModel/remoteMembers/Connection.ts b/src/state/CallViewModel/remoteMembers/Connection.ts index f320e6303..9310c9b82 100644 --- a/src/state/CallViewModel/remoteMembers/Connection.ts +++ b/src/state/CallViewModel/remoteMembers/Connection.ts @@ -13,8 +13,14 @@ import { import { ConnectionError, ConnectionErrorReason, + type Participant, type RemoteParticipant, + type RemoteTrackPublication, type Room as LivekitRoom, + RoomEvent, + type SubscriptionError, + type Track, + type TrackPublication, } from "livekit-client"; import { type LivekitTransportConfig } from "matrix-js-sdk/lib/matrixrtc"; import { BehaviorSubject, map } from "rxjs"; @@ -164,6 +170,7 @@ export class Connection { // Only tracks remote participants connectedParticipantsObserver(this.livekitRoom), ); + this.logRemoteTrackEvents(); scope.onEnd(() => { this.logger.info(`Connection scope ended, stopping connection`); @@ -171,6 +178,87 @@ export class Connection { }); } + /** + * Logs the lifecycle of remote participants and their tracks as seen by + * livekit-client, so that rageshakes show what a tile should have been + * rendering (published, subscribed, muted, paused) for each remote member. + */ + private logRemoteTrackEvents(): void { + const room = this.livekitRoom; + const log = this.logger.getChild("[RemoteTracks]"); + const track = (pub: TrackPublication, p: Participant): string => + `${pub.kind} ${pub.source} ${pub.trackSid} of ${p.identity}`; + + const onParticipantConnected = (p: RemoteParticipant): void => + log.info(`Participant connected: ${p.identity} (${p.sid})`); + const onParticipantDisconnected = (p: RemoteParticipant): void => + log.info(`Participant disconnected: ${p.identity} (${p.sid})`); + const onTrackPublished = ( + pub: RemoteTrackPublication, + p: RemoteParticipant, + ): void => log.info(`Published: ${track(pub, p)} muted=${pub.isMuted}`); + const onTrackUnpublished = ( + pub: RemoteTrackPublication, + p: RemoteParticipant, + ): void => log.info(`Unpublished: ${track(pub, p)}`); + const onTrackSubscribed = ( + _t: Track, + pub: RemoteTrackPublication, + p: RemoteParticipant, + ): void => log.info(`Subscribed: ${track(pub, p)} muted=${pub.isMuted}`); + const onTrackUnsubscribed = ( + _t: Track, + pub: RemoteTrackPublication, + p: RemoteParticipant, + ): void => log.info(`Unsubscribed: ${track(pub, p)}`); + const onTrackSubscriptionFailed = ( + trackSid: string, + p: RemoteParticipant, + reason?: SubscriptionError, + ): void => + log.warn( + `Subscription failed: ${trackSid} of ${p.identity} reason=${reason}`, + ); + // Mute events also fire for the local participant, which the Publisher + // already logs. + const onTrackMuted = (pub: TrackPublication, p: Participant): void => { + if (!p.isLocal) log.info(`Muted: ${track(pub, p)}`); + }; + const onTrackUnmuted = (pub: TrackPublication, p: Participant): void => { + if (!p.isLocal) log.info(`Unmuted: ${track(pub, p)}`); + }; + const onTrackStreamStateChanged = ( + pub: RemoteTrackPublication, + state: Track.StreamState, + p: RemoteParticipant, + ): void => log.info(`Stream ${state}: ${track(pub, p)}`); + + room + .on(RoomEvent.ParticipantConnected, onParticipantConnected) + .on(RoomEvent.ParticipantDisconnected, onParticipantDisconnected) + .on(RoomEvent.TrackPublished, onTrackPublished) + .on(RoomEvent.TrackUnpublished, onTrackUnpublished) + .on(RoomEvent.TrackSubscribed, onTrackSubscribed) + .on(RoomEvent.TrackUnsubscribed, onTrackUnsubscribed) + .on(RoomEvent.TrackSubscriptionFailed, onTrackSubscriptionFailed) + .on(RoomEvent.TrackMuted, onTrackMuted) + .on(RoomEvent.TrackUnmuted, onTrackUnmuted) + .on(RoomEvent.TrackStreamStateChanged, onTrackStreamStateChanged); + this.scope.onEnd(() => { + room + .off(RoomEvent.ParticipantConnected, onParticipantConnected) + .off(RoomEvent.ParticipantDisconnected, onParticipantDisconnected) + .off(RoomEvent.TrackPublished, onTrackPublished) + .off(RoomEvent.TrackUnpublished, onTrackUnpublished) + .off(RoomEvent.TrackSubscribed, onTrackSubscribed) + .off(RoomEvent.TrackUnsubscribed, onTrackUnsubscribed) + .off(RoomEvent.TrackSubscriptionFailed, onTrackSubscriptionFailed) + .off(RoomEvent.TrackMuted, onTrackMuted) + .off(RoomEvent.TrackUnmuted, onTrackUnmuted) + .off(RoomEvent.TrackStreamStateChanged, onTrackStreamStateChanged); + }); + } + /** * Starts the connection. * From 0b95fd5a65a87b03a4d52e1d97f5b3cc622adbc4 Mon Sep 17 00:00:00 2001 From: Matthew Hodgson Date: Thu, 3 Sep 2026 17:07:08 +0100 Subject: [PATCH 2/6] Log remote member participant matching and tile waiting state A tile shows "Waiting for media" for as long as its MatrixRTC member cannot be matched to a LiveKit participant. Log each transition of that match and of the tile's waitingForMedia state so rageshakes can tie a stuck tile to the LiveKit participant and track events. --- .../MatrixLivekitMembers.test.ts | 7 ++++- .../remoteMembers/MatrixLivekitMembers.ts | 10 ++++++- src/state/media/RemoteUserMediaViewModel.ts | 26 ++++++++++++------- 3 files changed, 31 insertions(+), 12 deletions(-) diff --git a/src/state/CallViewModel/remoteMembers/MatrixLivekitMembers.test.ts b/src/state/CallViewModel/remoteMembers/MatrixLivekitMembers.test.ts index 244d70ae8..09c827896 100644 --- a/src/state/CallViewModel/remoteMembers/MatrixLivekitMembers.test.ts +++ b/src/state/CallViewModel/remoteMembers/MatrixLivekitMembers.test.ts @@ -5,7 +5,8 @@ SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial Please see LICENSE in the repository root for full details. */ -import { describe, test, expect, beforeEach, afterEach } from "vitest"; +import { describe, test, expect, beforeEach, afterEach, vi } from "vitest"; +import { logger } from "matrix-js-sdk/lib/logger"; import { type CallMembership, type LivekitTransport, @@ -77,6 +78,7 @@ function epochMeWith$( } test("should signal participant not yet connected to livekit", async () => { + const info = vi.spyOn(logger, "info"); const mockedMemberships$ = new BehaviorSubject([bobMembership]); const mockConnectionManagerData$ = new BehaviorSubject( new ConnectionManagerData(), @@ -108,6 +110,9 @@ test("should signal participant not yet connected to livekit", async () => { return true; }, ); + expect(info).toHaveBeenCalledWith( + "[RemoteMatrixLivekitMembers] @bob:example.org:DEV000: LiveKit participant missing", + ); }); // Helper to create epoch'ed memberships$ and membershipsWithTransport$ from memberships observable. diff --git a/src/state/CallViewModel/remoteMembers/MatrixLivekitMembers.ts b/src/state/CallViewModel/remoteMembers/MatrixLivekitMembers.ts index 0b93a274b..e884d3821 100644 --- a/src/state/CallViewModel/remoteMembers/MatrixLivekitMembers.ts +++ b/src/state/CallViewModel/remoteMembers/MatrixLivekitMembers.ts @@ -11,6 +11,7 @@ import { type LivekitTransportConfig, } from "matrix-js-sdk/lib/matrixrtc"; import { combineLatest, filter, map } from "rxjs"; +import { logger } from "matrix-js-sdk/lib/logger"; import { type Behavior } from "../../Behavior"; import { type IConnectionManager } from "./ConnectionManager"; @@ -133,8 +134,15 @@ export function createRemoteMatrixLivekitMembers$({ } }, // Each update where the key of the generator array do not change will result in updates to the `data$` behavior. - (scope, data$, userId, _deviceId, _memberId, _rtcBackendIdentity) => { + (scope, data$, userId, _deviceId, _memberId, rtcBackendIdentity) => { const { participant$, ...rest } = scope.splitBehavior(data$); + // 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) => { + logger.info( + `[RemoteMatrixLivekitMembers] ${rtcBackendIdentity}: LiveKit participant ${p ? `matched (${p.sid})` : "missing"}`, + ); + }); // will only get called once per backend identity. // updates to data$ and as a result to displayName$ and mxcAvatarUrl$ are more frequent. return { diff --git a/src/state/media/RemoteUserMediaViewModel.ts b/src/state/media/RemoteUserMediaViewModel.ts index 4307dea41..6daa381a7 100644 --- a/src/state/media/RemoteUserMediaViewModel.ts +++ b/src/state/media/RemoteUserMediaViewModel.ts @@ -8,6 +8,7 @@ Please see LICENSE in the repository root for full details. import { type RemoteParticipant } from "livekit-client"; import { combineLatest, map, of, switchMap } from "rxjs"; +import { logger } from "matrix-js-sdk/lib/logger"; import { type Behavior } from "../Behavior"; import { createVolumeControls, type VolumeControls } from "../VolumeControls"; @@ -45,6 +46,20 @@ export function createRemoteUserMedia( statsType: "inbound-rtp", }); + const waitingForMedia$ = scope.behavior( + combineLatest( + [inputs.livekitRoom$, inputs.participant$], + (livekitRoom, participant) => + // If livekitRoom is undefined, the user is not attempting to publish on + // any transport and so we shouldn't expect a participant. (They might + // be a subscribe-only bot for example.) + livekitRoom !== undefined && participant === null, + ), + ); + waitingForMedia$.pipe(scope.bind()).subscribe((waiting) => { + logger.info(`[RemoteUserMedia ${inputs.id}] waitingForMedia=${waiting}`); + }); + return { ...baseUserMedia, ...createVolumeControls(scope, { @@ -68,15 +83,6 @@ export function createRemoteUserMedia( ), ), ), - waitingForMedia$: scope.behavior( - combineLatest( - [inputs.livekitRoom$, inputs.participant$], - (livekitRoom, participant) => - // If livekitRoom is undefined, the user is not attempting to publish on - // any transport and so we shouldn't expect a participant. (They might - // be a subscribe-only bot for example.) - livekitRoom !== undefined && participant === null, - ), - ), + waitingForMedia$, }; } From 2ab2197c01d23d51bd3a6ade455f7b14c919bdf3 Mon Sep 17 00:00:00 2001 From: Matthew Hodgson Date: Thu, 3 Sep 2026 17:26:38 +0100 Subject: [PATCH 3/6] Warn when a behavior is re-entered while delivering a value rxjs delivers a nested emission to every subscriber and then resumes delivering the outer, older value to the remaining subscribers, which leaves them permanently out of sync. Log the first occurrence per behavior with a stack trace so the re-entrant path can be identified from a rageshake. --- src/state/ObservableScope.test.ts | 27 +++++++++++++++++++++++++++ src/state/ObservableScope.ts | 24 +++++++++++++++++++++++- 2 files changed, 50 insertions(+), 1 deletion(-) diff --git a/src/state/ObservableScope.test.ts b/src/state/ObservableScope.test.ts index 31728f394..353ecaed5 100644 --- a/src/state/ObservableScope.test.ts +++ b/src/state/ObservableScope.test.ts @@ -237,3 +237,30 @@ describe("Reconcile", () => { expect(setup).toHaveBeenCalledWith(1); }); }); + +describe("behavior", () => { + it("warns when a subscriber re-enters the behavior synchronously", () => { + const warn = vi.spyOn(logger, "warn").mockImplementation(() => {}); + const scope = new ObservableScope(); + const source$ = new Subject(); + const behavior$ = scope.behavior(source$, 0); + // A subscriber that reacts to the value 1 by synchronously emitting 2 + behavior$.subscribe((v) => { + if (v === 1) source$.next(2); + }); + const seen: number[] = []; + behavior$.subscribe((v) => seen.push(v)); + + source$.next(1); + + expect(warn).toHaveBeenCalledWith( + expect.stringContaining("Behavior re-entered"), + expect.any(String), + ); + // Documents the hazard the warning is about: the later subscriber ends up + // with the stale value 1 even though the behavior's value is 2. + expect(behavior$.value).toBe(2); + expect(seen.at(-1)).toBe(1); + scope.end(); + }); +}); diff --git a/src/state/ObservableScope.ts b/src/state/ObservableScope.ts index e3fc644f7..0a64e2a42 100644 --- a/src/state/ObservableScope.ts +++ b/src/state/ObservableScope.ts @@ -20,6 +20,8 @@ import { takeUntil, } from "rxjs"; +import { logger } from "matrix-js-sdk/lib/logger"; + import { type Behavior } from "./Behavior"; type MonoTypeOperator = (o: Observable) => Observable; @@ -73,9 +75,29 @@ export class ObservableScope { // they will no longer re-emit their current value upon subscription. We want // to support Observables that complete (for example `of({})`), so we have to // take care to not propagate the completion event. + // If a subscriber synchronously causes this same behavior to emit again, + // rxjs delivers the nested value to every subscriber first and then + // resumes delivering the outer (older) value to the remaining subscribers, + // leaving them permanently out of sync with the others. Log the first + // occurrence with a stack trace so that the re-entrant path can be found. + let delivering = false; + let reentryReported = false; setValue$.pipe(this.bind(), distinctUntilChanged()).subscribe({ next(value) { - subject$.next(value); + if (delivering && !reentryReported) { + reentryReported = true; + logger.warn( + "Behavior re-entered while delivering a value; later subscribers will be left with a stale value", + new Error().stack, + ); + } + const wasDelivering = delivering; + delivering = true; + try { + subject$.next(value); + } finally { + delivering = wasDelivering; + } }, error(err: unknown) { subject$.error(err); From b1393998a7f7282cb7f4a4727edc2e91bca61378 Mon Sep 17 00:00:00 2001 From: Matthew Hodgson Date: Thu, 3 Sep 2026 18:08:53 +0100 Subject: [PATCH 4/6] Report re-entrancy depth and tag torn behaviors A single re-entrant emission can only strand a contiguous run of a behavior's subscribers, so a tile that is both 'speaking' and 'waiting for media' (a middle subscriber stranded) implies a nested re-entry. Report the nesting depth and log each new depth once instead of only the first re-entry, and tag splitBehavior-derived behaviors with their field name so the warning identifies which behavior tore. --- src/state/ObservableScope.test.ts | 26 ++++++++++++++---------- src/state/ObservableScope.ts | 33 ++++++++++++++++++++----------- 2 files changed, 38 insertions(+), 21 deletions(-) diff --git a/src/state/ObservableScope.test.ts b/src/state/ObservableScope.test.ts index 353ecaed5..24b5d268d 100644 --- a/src/state/ObservableScope.test.ts +++ b/src/state/ObservableScope.test.ts @@ -239,27 +239,33 @@ describe("Reconcile", () => { }); describe("behavior", () => { - it("warns when a subscriber re-enters the behavior synchronously", () => { + it("warns with the tag and nesting depth, logging each new depth once", () => { const warn = vi.spyOn(logger, "warn").mockImplementation(() => {}); const scope = new ObservableScope(); const source$ = new Subject(); - const behavior$ = scope.behavior(source$, 0); - // A subscriber that reacts to the value 1 by synchronously emitting 2 + const behavior$ = scope.behavior(source$, 0, "tagged"); + // Re-enter once on value 1 (depth 2), then a nested re-entry on value 2 + // (depth 3), which is the shape needed to strand a middle subscriber. behavior$.subscribe((v) => { if (v === 1) source$.next(2); }); + behavior$.subscribe((v) => { + if (v === 2) source$.next(3); + }); const seen: number[] = []; behavior$.subscribe((v) => seen.push(v)); source$.next(1); - expect(warn).toHaveBeenCalledWith( - expect.stringContaining("Behavior re-entered"), - expect.any(String), - ); - // Documents the hazard the warning is about: the later subscriber ends up - // with the stale value 1 even though the behavior's value is 2. - expect(behavior$.value).toBe(2); + const messages = warn.mock.calls.map((c) => c[0] as string); + expect(messages).toEqual([ + expect.stringContaining("Behavior (tagged) re-entered at depth 2"), + expect.stringContaining("Behavior (tagged) re-entered at depth 3"), + ]); + warn.mock.calls.forEach((c) => expect(c[1]).toEqual(expect.any(String))); + // The behavior settles on the newest value while the last subscriber is + // left stranded on the oldest. + expect(behavior$.value).toBe(3); expect(seen.at(-1)).toBe(1); scope.end(); }); diff --git a/src/state/ObservableScope.ts b/src/state/ObservableScope.ts index 0a64e2a42..d0cfb7e93 100644 --- a/src/state/ObservableScope.ts +++ b/src/state/ObservableScope.ts @@ -68,6 +68,9 @@ export class ObservableScope { public behavior( setValue$: Observable, initialValue: T | typeof nothing = nothing, + // An optional label used only in the re-entrancy warning below, so that a + // torn behavior can be identified from a rageshake. + tag?: string, ): Behavior { const subject$ = new BehaviorSubject(initialValue); // Push values from the Observable into the BehaviorSubject. @@ -78,25 +81,29 @@ export class ObservableScope { // If a subscriber synchronously causes this same behavior to emit again, // rxjs delivers the nested value to every subscriber first and then // resumes delivering the outer (older) value to the remaining subscribers, - // leaving them permanently out of sync with the others. Log the first - // occurrence with a stack trace so that the re-entrant path can be found. - let delivering = false; - let reentryReported = false; + // leaving them permanently out of sync with the others. A single such + // re-entry can only strand a contiguous run of subscribers; stranding a + // subscriber in the middle of the list needs a nested (deeper) re-entry, so + // we report the depth and log each new depth (with a stack trace) rather + // than only the first occurrence, to make a nested re-entry visible. + let depth = 0; + let maxReportedDepth = 1; setValue$.pipe(this.bind(), distinctUntilChanged()).subscribe({ next(value) { - if (delivering && !reentryReported) { - reentryReported = true; + if (depth > 0 && depth + 1 > maxReportedDepth) { + maxReportedDepth = depth + 1; logger.warn( - "Behavior re-entered while delivering a value; later subscribers will be left with a stale value", + `Behavior${tag ? ` (${tag})` : ""} re-entered at depth ${ + depth + 1 + } while delivering a value; later subscribers will be left with a stale value`, new Error().stack, ); } - const wasDelivering = delivering; - delivering = true; + depth++; try { subject$.next(value); } finally { - delivering = wasDelivering; + depth--; } }, error(err: unknown) { @@ -193,7 +200,11 @@ export class ObservableScope { return Object.fromEntries( Object.keys(input$.value).map((key) => [ `${key}$`, - this.behavior(input$.pipe(map((input) => input[key as keyof T]))), + this.behavior( + input$.pipe(map((input) => input[key as keyof T])), + nothing, + key, + ), ]), ) as SplitBehavior; } From dee4b05695dc9ead77c912d6dec7e3310e3963c7 Mon Sep 17 00:00:00 2001 From: Matthew Hodgson Date: Thu, 3 Sep 2026 18:43:18 +0100 Subject: [PATCH 5/6] Diagnostics for duplicate participant matches and mid-delivery scope teardown Warn when ConnectionManagerData merges participants from a second connection to the same SFU URL, when more than one LiveKit participant matches a member's backend identity, and when a scope is ended while one of its behaviors is still delivering a value (which strands the later bound subscribers on the old value). The per-member match log now also names the connection the participant came from. --- .../remoteMembers/ConnectionManager.test.ts | 20 ++++++++++++++++- .../remoteMembers/ConnectionManager.ts | 9 ++++++-- .../MatrixLivekitMembers.test.ts | 2 +- .../remoteMembers/MatrixLivekitMembers.ts | 15 ++++++++----- src/state/ObservableScope.test.ts | 22 +++++++++++++++++++ src/state/ObservableScope.ts | 17 ++++++++++++-- 6 files changed, 74 insertions(+), 11 deletions(-) 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); } From 054913d17d1ad8e13b0aec221c0a9a2d9f315153 Mon Sep 17 00:00:00 2001 From: Robin Date: Thu, 3 Sep 2026 20:11:18 +0200 Subject: [PATCH 6/6] Use testScope in ObservableScope tests --- src/state/ObservableScope.test.ts | 13 ++++++------- 1 file changed, 6 insertions(+), 7 deletions(-) diff --git a/src/state/ObservableScope.test.ts b/src/state/ObservableScope.test.ts index 12bd39ced..25bb14324 100644 --- a/src/state/ObservableScope.test.ts +++ b/src/state/ObservableScope.test.ts @@ -16,7 +16,7 @@ import { ObservableScope, trackEpoch, } from "./ObservableScope"; -import { withTestScheduler } from "../utils/test"; +import { testScope, withTestScheduler } from "../utils/test"; describe("Epoch", () => { it("should map the value correctly", () => { @@ -87,7 +87,7 @@ describe("Epoch", () => { }); it("behavior test", () => { - const scope = new ObservableScope(); + const scope = testScope(); const s$ = new Subject(); const behavior$ = scope.behavior(s$, 0); behavior$.subscribe((value) => { @@ -115,7 +115,7 @@ describe("Reconcile", () => { it("should wait clean up before processing next", async () => { vi.useFakeTimers(); - const scope = new ObservableScope(); + const scope = testScope(); const behavior$ = new BehaviorSubject(0); const setup = vi.fn().mockImplementation(async () => await sleep(100)); @@ -149,7 +149,7 @@ describe("Reconcile", () => { it("should skip intermediates values that are not setup", async () => { vi.useFakeTimers(); - const scope = new ObservableScope(); + const scope = testScope(); const behavior$ = new BehaviorSubject(0); const setup = vi @@ -194,7 +194,7 @@ describe("Reconcile", () => { it("should wait for setup to complete before starting cleanup", async () => { vi.useFakeTimers(); - const scope = new ObservableScope(); + const scope = testScope(); const behavior$ = new BehaviorSubject(0); const setup = vi @@ -241,7 +241,7 @@ describe("Reconcile", () => { describe("behavior", () => { it("warns with the tag and nesting depth, logging each new depth once", () => { const warn = vi.spyOn(logger, "warn").mockImplementation(() => {}); - const scope = new ObservableScope(); + const scope = testScope(); const source$ = new Subject(); const behavior$ = scope.behavior(source$, 0, "tagged"); // Re-enter once on value 1 (depth 2), then a nested re-entry on value 2 @@ -267,7 +267,6 @@ describe("behavior", () => { // left stranded on the oldest. expect(behavior$.value).toBe(3); expect(seen.at(-1)).toBe(1); - scope.end(); }); it("warns when the scope is ended while a behavior is mid-delivery", () => {