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.
This commit is contained in:
Matthew Hodgson
2026-09-03 18:43:18 +01:00
parent b1393998a7
commit dee4b05695
6 changed files with 74 additions and 11 deletions
@@ -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<string, Behavior<RemoteParticipant[]>>;
@@ -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,
),
),
@@ -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",
);
});
@@ -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.