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. * 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 244d70ae8..fe7621121 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 on no connection", + ); }); // 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..76f4670ec 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"; @@ -110,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. @@ -133,8 +138,16 @@ 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) => { + const url = data$.value.connection?.transport.livekit_service_url; + logger.info( + `[RemoteMatrixLivekitMembers] ${rtcBackendIdentity}: LiveKit participant ${p ? `matched (${p.sid})` : "missing"} on ${url ?? "no connection"}`, + ); + }); // 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/ObservableScope.test.ts b/src/state/ObservableScope.test.ts index 31728f394..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 @@ -237,3 +237,57 @@ describe("Reconcile", () => { expect(setup).toHaveBeenCalledWith(1); }); }); + +describe("behavior", () => { + it("warns with the tag and nesting depth, logging each new depth once", () => { + const warn = vi.spyOn(logger, "warn").mockImplementation(() => {}); + 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 + // (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); + + 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); + }); + + 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 e3fc644f7..8f09cb6ef 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; @@ -34,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( @@ -66,6 +74,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. @@ -73,11 +84,37 @@ 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. 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) { - subject$.next(value); + next: (value) => { + if (depth > 0 && depth + 1 > maxReportedDepth) { + maxReportedDepth = depth + 1; + logger.warn( + `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, + ); + } + depth++; + this.delivering++; + try { + subject$.next(value); + } finally { + depth--; + this.delivering--; + } }, - error(err: unknown) { + error: (err: unknown) => { subject$.error(err); }, }); @@ -90,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); } @@ -171,7 +213,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; } 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$, }; }