Merge upstream/livekit into branch

This commit is contained in:
Ryan Emmick
2026-06-11 22:11:21 -05:00
191 changed files with 13413 additions and 4649 deletions

View File

@@ -13,6 +13,7 @@ import { MembershipManagerEvent, Status } from "matrix-js-sdk/lib/matrixrtc";
import { ObservableScope } from "../../ObservableScope";
import { createHomeserverConnected$ } from "./HomeserverConnected";
import { testScope, withTestScheduler } from "../../../utils/test";
/**
* Minimal stub of a Matrix client sufficient for our tests:
@@ -96,107 +97,240 @@ describe("createHomeserverConnected$", () => {
// LLM generated test cases. They are a bit overkill but I improved the mocking so it is
// easy enough to read them so I think they can stay.
it("is false when sync state is not Syncing", () => {
const hsConnected = createHomeserverConnected$(scope, client, session);
expect(hsConnected.combined$.value).toBe(false);
// Note: gracePeriodMs is set to 0 to avoid debouncing delays in tests
it("reports syncing reason when sync state is not Syncing", () => {
const hsConnected = createHomeserverConnected$(scope, client, session, 0);
expect(hsConnected.combined$.value).toEqual([false, "sync"]);
});
it("remains false while membership status is not Connected even if sync is Syncing", () => {
const hsConnected = createHomeserverConnected$(scope, client, session);
it("reports membership reason when sync is Syncing but membership is not Connected", () => {
const hsConnected = createHomeserverConnected$(scope, client, session, 0);
client.setSyncState(SyncState.Syncing);
expect(hsConnected.combined$.value).toBe(false); // membership still disconnected
expect(hsConnected.combined$.value).toEqual([false, "membership"]);
});
it("is false when membership status transitions to Connected but ProbablyLeft is true", () => {
const hsConnected = createHomeserverConnected$(scope, client, session);
it("reports probablyLeft reason when membership transitions to Connected but ProbablyLeft is true", () => {
const hsConnected = createHomeserverConnected$(scope, client, session, 0);
// Make sync loop OK
client.setSyncState(SyncState.Syncing);
// Indicate probable leave before connection
session.setProbablyLeft(true);
session.setMembershipStatus(Status.Connected);
expect(hsConnected.combined$.value).toBe(false);
expect(hsConnected.combined$.value).toEqual([false, "probablyLeft"]);
});
it("becomes true only when all three conditions are satisfied", () => {
const hsConnected = createHomeserverConnected$(scope, client, session);
it("becomes null (connected) only when all three conditions are satisfied", () => {
const hsConnected = createHomeserverConnected$(scope, client, session, 0);
// 1. Sync loop connected
client.setSyncState(SyncState.Syncing);
expect(hsConnected.combined$.value).toBe(false); // not yet membership connected
expect(hsConnected.combined$.value).toEqual([false, "membership"]); // not yet membership connected
// 2. Membership connected
session.setMembershipStatus(Status.Connected);
expect(hsConnected.combined$.value).toBe(true); // probablyLeft is false
expect(hsConnected.combined$.value).toEqual([true, null]); // probablyLeft is false
});
it("drops back to false when sync loop leaves Syncing", () => {
const hsConnected = createHomeserverConnected$(scope, client, session);
it("returns syncing reason when sync loop leaves Syncing", () => {
const hsConnected = createHomeserverConnected$(scope, client, session, 0);
// Reach connected state
client.setSyncState(SyncState.Syncing);
session.setMembershipStatus(Status.Connected);
expect(hsConnected.combined$.value).toBe(true);
expect(hsConnected.combined$.value).toEqual([true, null]);
// Sync loop error => should flip false
// Sync loop error => should report syncing reason
client.setSyncState(SyncState.Error);
expect(hsConnected.combined$.value).toBe(false);
expect(hsConnected.combined$.value).toEqual([false, "sync"]);
});
it("drops back to false when membership status becomes disconnected", () => {
const hsConnected = createHomeserverConnected$(scope, client, session);
it("returns membershipConnected reason when membership status becomes disconnected", () => {
const hsConnected = createHomeserverConnected$(scope, client, session, 0);
client.setSyncState(SyncState.Syncing);
session.setMembershipStatus(Status.Connected);
expect(hsConnected.combined$.value).toBe(true);
expect(hsConnected.combined$.value).toEqual([true, null]);
session.setMembershipStatus(Status.Disconnected);
expect(hsConnected.combined$.value).toBe(false);
expect(hsConnected.combined$.value).toEqual([false, "membership"]);
});
it("drops to false when ProbablyLeft is emitted after being true", () => {
const hsConnected = createHomeserverConnected$(scope, client, session);
it("returns certainlyConnected reason when ProbablyLeft is emitted", () => {
const hsConnected = createHomeserverConnected$(scope, client, session, 0);
client.setSyncState(SyncState.Syncing);
session.setMembershipStatus(Status.Connected);
expect(hsConnected.combined$.value).toBe(true);
expect(hsConnected.combined$.value).toEqual([true, null]);
session.setProbablyLeft(true);
expect(hsConnected.combined$.value).toBe(false);
expect(hsConnected.combined$.value).toEqual([false, "probablyLeft"]);
});
it("recovers to true if ProbablyLeft becomes false again while other conditions remain true", () => {
const hsConnected = createHomeserverConnected$(scope, client, session);
it("recovers to null (connected) if ProbablyLeft becomes false again while other conditions remain true", () => {
const hsConnected = createHomeserverConnected$(scope, client, session, 0);
client.setSyncState(SyncState.Syncing);
session.setMembershipStatus(Status.Connected);
expect(hsConnected.combined$.value).toBe(true);
expect(hsConnected.combined$.value).toEqual([true, null]);
session.setProbablyLeft(true);
expect(hsConnected.combined$.value).toBe(false);
expect(hsConnected.combined$.value).toEqual([false, "probablyLeft"]);
// Simulate clearing the flag (in realistic scenario membership manager would update)
session.setProbablyLeft(false);
expect(hsConnected.combined$.value).toBe(true);
expect(hsConnected.combined$.value).toEqual([true, null]);
});
it("composite sequence reflects each individual failure reason", () => {
const hsConnected = createHomeserverConnected$(scope, client, session);
const hsConnected = createHomeserverConnected$(scope, client, session, 0);
// Initially false (sync error + disconnected + not probably left)
expect(hsConnected.combined$.value).toBe(false);
// Initially: sync error + membership disconnected → syncing wins (highest priority)
expect(hsConnected.combined$.value).toEqual([false, "sync"]);
// Fix sync only
// Fix sync only → membershipConnected is now the blocker
client.setSyncState(SyncState.Syncing);
expect(hsConnected.combined$.value).toBe(false);
expect(hsConnected.combined$.value).toEqual([false, "membership"]);
// Fix membership
// Fix membership → all conditions satisfied
session.setMembershipStatus(Status.Connected);
expect(hsConnected.combined$.value).toBe(true);
expect(hsConnected.combined$.value).toEqual([true, null]);
// Introduce probablyLeft -> false
// Introduce probablyLeft → certainlyConnected
session.setProbablyLeft(true);
expect(hsConnected.combined$.value).toBe(false);
expect(hsConnected.combined$.value).toEqual([false, "probablyLeft"]);
// Restore notProbablyLeft -> true again
// Restore notProbablyLeft → connected again
session.setProbablyLeft(false);
expect(hsConnected.combined$.value).toBe(true);
expect(hsConnected.combined$.value).toEqual([true, null]);
// Drop sync -> false
// Drop sync → syncing reason
client.setSyncState(SyncState.Error);
expect(hsConnected.combined$.value).toBe(false);
expect(hsConnected.combined$.value).toEqual([false, "sync"]);
});
});
describe("createHomeserverConnected$ - combined$ reason values", () => {
let scope: ObservableScope;
let client: MockMatrixClient;
let session: MockMatrixRTCSession;
beforeEach(() => {
scope = new ObservableScope();
// Start with sync failing and membership disconnected
client = new MockMatrixClient(SyncState.Error);
session = new MockMatrixRTCSession({
membershipStatus: Status.Disconnected,
probablyLeft: false,
});
});
afterEach(() => {
scope.end();
});
it("is [true, null] when all three conditions are satisfied", () => {
const { combined$ } = createHomeserverConnected$(scope, client, session, 0);
client.setSyncState(SyncState.Syncing);
session.setMembershipStatus(Status.Connected);
expect(combined$.value).toEqual([true, null]);
});
it("reports syncing when sync loop is not Syncing", () => {
const { combined$ } = createHomeserverConnected$(scope, client, session, 0);
// client starts with SyncState.Error, membership also disconnected
expect(combined$.value).toEqual([false, "sync"]);
});
it("reports membershipConnected when sync is fine but membership is not Connected", () => {
const { combined$ } = createHomeserverConnected$(scope, client, session, 0);
client.setSyncState(SyncState.Syncing);
// session still Status.Disconnected
expect(combined$.value).toEqual([false, "membership"]);
});
it("reports certainlyConnected when probablyLeft is true", () => {
const { combined$ } = createHomeserverConnected$(scope, client, session, 0);
client.setSyncState(SyncState.Syncing);
session.setMembershipStatus(Status.Connected);
session.setProbablyLeft(true);
expect(combined$.value).toEqual([false, "probablyLeft"]);
});
it("prioritises syncing over membershipConnected when both fail", () => {
const { combined$ } = createHomeserverConnected$(scope, client, session, 0);
// Both sync (Error) and membership (Disconnected) are failing
expect(combined$.value).toEqual([false, "sync"]);
});
it("updates reason as conditions change", () => {
const { combined$ } = createHomeserverConnected$(scope, client, session, 0);
// Initially: syncing fails
expect(combined$.value).toEqual([false, "sync"]);
// Fix sync → membershipConnected is now the blocker
client.setSyncState(SyncState.Syncing);
expect(combined$.value).toEqual([false, "membership"]);
// Fix membership → probablyLeft makes certainlyConnected fail
session.setProbablyLeft(true);
session.setMembershipStatus(Status.Connected);
expect(combined$.value).toEqual([false, "probablyLeft"]);
// Clear probablyLeft → all conditions satisfied
session.setProbablyLeft(false);
expect(combined$.value).toEqual([true, null]);
});
});
describe("createHomeserverConnected$ - Grace Period", () => {
const GRACE_PERIOD = 5;
function marbleTest(
syncStateMarbles: string,
expectedConnectedMarbles: string,
): void {
withTestScheduler(({ behavior, schedule, expectObservable }) => {
const syncState$ = behavior(syncStateMarbles, {
s: SyncState.Syncing,
e: SyncState.Error,
});
const client = new MockMatrixClient(syncState$.value);
schedule(syncStateMarbles, {
s: () => client.setSyncState(SyncState.Syncing),
e: () => client.setSyncState(SyncState.Error),
});
const session = new MockMatrixRTCSession({
membershipStatus: Status.Connected,
probablyLeft: false,
});
const hsConnected = createHomeserverConnected$(
testScope(),
client,
session,
GRACE_PERIOD,
);
expectObservable(hsConnected.combined$).toBe(expectedConnectedMarbles, {
y: [true, null],
n: [false, "sync"],
});
});
}
it("respects gracePeriodMs: stays true during grace period and flips false after", () => {
// - Initial state: Everything is connected
// - Sync error occurs -> should remain connected due to grace period
// - After grace period, not connected
marbleTest("se", "y-----n");
// If the sync error takes longer to occur, it should take equally long for
// the connection state to change
marbleTest("s--e", "y-------n");
});
it("recovers immediately if sync returns during grace period", () => {
// - Initial state: Connected
// - Sync error occurs
// - Sync recovers BEFORE the grace period expires
// - Connection state remains constant
marbleTest("se--s", "y");
});
it("flips to true IMMEDIATELY even if a grace period was pending", () => {
// - Initial error: connection eventually flips to false
// - Back to Syncing -> Must be connected immediately (synchronously)
marbleTest("e-----s", "y----ny");
});
});

View File

@@ -12,12 +12,23 @@ import {
type MatrixRTCSession,
} from "matrix-js-sdk/lib/matrixrtc";
import { ClientEvent, type MatrixClient, SyncState } from "matrix-js-sdk";
import { fromEvent, startWith, map, tap, type Observable } from "rxjs";
import {
fromEvent,
startWith,
map,
tap,
type Observable,
distinctUntilChanged,
switchMap,
of,
delay,
combineLatest,
} from "rxjs";
import { logger as rootLogger } from "matrix-js-sdk/lib/logger";
import { Config } from "../../../config/Config";
import { type ObservableScope } from "../../ObservableScope";
import { type Behavior } from "../../Behavior";
import { and$ } from "../../../utils/observable";
import { type NodeStyleEventEmitter } from "../../../utils/test";
/**
@@ -25,8 +36,14 @@ import { type NodeStyleEventEmitter } from "../../../utils/test";
*/
const logger = rootLogger.getChild("[HomeserverConnected]");
export type HomeserverDisconnectReason = "sync" | "membership" | "probablyLeft";
export interface HomeserverConnected {
combined$: Behavior<boolean>;
/**
* Emits `[true, null]` when the homeserver connection is healthy, or
* `[false, reason]` when one of the three sub-conditions fails.
*/
combined$: Behavior<[boolean, HomeserverDisconnectReason | null]>;
rtsSession$: Behavior<Status>;
}
@@ -34,29 +51,48 @@ export interface HomeserverConnected {
* Behavior representing whether we consider ourselves connected to the Matrix homeserver
* for the purposes of a MatrixRTC session.
*
* Becomes FALSE if ANY sub-condition is fulfilled:
* 1. Sync loop is not in SyncState.Syncing
* 2. membershipStatus !== Status.Connected
* 3. probablyLeft === true
* `combined$` emits `null` when all conditions are satisfied, or the first failing
* reason (priority: syncing > membershipConnected > certainlyConnected):
* 1. Sync loop is not in SyncState.Syncing (after grace period) → "sync"
* 2. membershipStatus !== Status.Connected → "membership"
* 3. probablyLeft === true → "probablyLeft"
*
* @param scope - The observable scope for lifecycle management.
* @param client - The Matrix client to monitor sync state.
* @param matrixRTCSession - The RTC session to monitor membership.
* @param gracePeriodMs - Grace period in milliseconds to wait before reporting sync disconnect.
* If not provided, uses the config value (default 10000ms).
*/
export function createHomeserverConnected$(
scope: ObservableScope,
client: NodeStyleEventEmitter & Pick<MatrixClient, "getSyncState">,
matrixRTCSession: NodeStyleEventEmitter &
Pick<MatrixRTCSession, "membershipStatus" | "probablyLeft">,
gracePeriodMs?: number,
): HomeserverConnected {
// Get grace period from parameter or config (default 10000ms)
const graceMs = gracePeriodMs ?? Config.get().sync_disconnect_grace_period_ms;
const syncing$ = (
fromEvent(client, ClientEvent.Sync) as Observable<[SyncState]>
).pipe(
startWith([client.getSyncState()]),
map(([state]) => state === SyncState.Syncing),
distinctUntilChanged(),
switchMap((isSyncing) => {
if (isSyncing || graceMs <= 0) {
return of(isSyncing);
}
return of(false).pipe(delay(graceMs), startWith(true));
}),
distinctUntilChanged(),
);
const rtsSession$ = scope.behavior<Status>(
fromEvent(matrixRTCSession, MembershipManagerEvent.StatusChanged).pipe(
map(() => matrixRTCSession.membershipStatus ?? Status.Unknown),
),
Status.Unknown,
matrixRTCSession.membershipStatus ?? Status.Unknown,
);
const membershipConnected$ = rtsSession$.pipe(
@@ -80,9 +116,22 @@ export function createHomeserverConnected$(
);
const combined$ = scope.behavior(
and$(syncing$, membershipConnected$, certainlyConnected$).pipe(
tap((connected) => {
logger.info(`Homeserver connected update: ${connected}`);
combineLatest([syncing$, membershipConnected$, certainlyConnected$]).pipe(
map(
([syncing, membership, certainly]): [
boolean,
HomeserverDisconnectReason | null,
] => {
if (!syncing) return [false, "sync"];
if (!membership) return [false, "membership"];
if (!certainly) return [false, "probablyLeft"];
return [true, null];
},
),
tap(([connected, reason]) => {
logger.info(
`Homeserver connected update: ${connected ? "connected" : reason}`,
);
}),
),
);

View File

@@ -11,13 +11,23 @@ import {
type LivekitTransportConfig,
type MatrixRTCSession,
} from "matrix-js-sdk/lib/matrixrtc";
import { describe, expect, it, vi } from "vitest";
import {
describe,
expect,
it,
vi,
beforeAll,
afterAll,
beforeEach,
} from "vitest";
import { AutoDiscovery } from "matrix-js-sdk/lib/autodiscovery";
import { BehaviorSubject, map, of } from "rxjs";
import { logger } from "matrix-js-sdk/lib/logger";
import { type LocalParticipant, type LocalTrack } from "livekit-client";
import { MatrixRTCMode } from "../../../settings/settings";
import { PosthogAnalytics } from "../../../analytics/PosthogAnalytics";
import { MatrixRTCMode } from "../../../config/ConfigOptions";
import { type HomeserverDisconnectReason } from "./HomeserverConnected";
import {
flushPromises,
mockConfig,
@@ -215,9 +225,13 @@ describe("LocalMembership", () => {
createPublisherFactory: vi.fn(),
joinMatrixRTC: async (): Promise<void> => {},
homeserverConnected: {
combined$: constant(true),
combined$: constant<[boolean, HomeserverDisconnectReason | null]>([
true,
null,
]),
rtsSession$: constant(RTCMemberStatus.Connected),
},
roomId: "!test-room-id:example.org",
};
it("throws error on missing RTC config error", () => {
@@ -667,4 +681,210 @@ describe("LocalMembership", () => {
// expect(publishers[0].stopTracks).toHaveBeenCalled();
});
// TODO add tests for matrix local matrix participation.
describe("reconnecting analytics", () => {
beforeAll(() => {
mockConfig();
});
beforeEach(() => {
vi.restoreAllMocks();
});
afterAll(() => {
PosthogAnalytics.resetInstance();
});
it("does not fire CallReconnecting for the initial non-connected state at startup", async () => {
const scope = new ObservableScope();
const trackSpy = vi.spyOn(
PosthogAnalytics.instance.eventCallReconnecting,
"track",
);
// Simulate startup where membership isn't established yet
const hsReason$ = new BehaviorSubject<
[boolean, HomeserverDisconnectReason | null]
>([false, "membership"]);
const connectionManagerData = new ConnectionManagerData();
connectionManagerData.add(connectionTransportAConnected, []);
createLocalMembership$({
scope,
...defaultCreateLocalMemberValues,
homeserverConnected: {
combined$: hsReason$,
rtsSession$: constant(RTCMemberStatus.Connected),
},
connectionManager: {
connectionManagerData$: constant(new Epoch(connectionManagerData)),
},
localTransport$: new BehaviorSubject({
advertised$: new BehaviorSubject(aTransport),
active$: new BehaviorSubject(aTransportWithSFUConfig),
}),
});
await flushPromises();
// Membership is established — call is now connected
hsReason$.next([true, null]);
expect(trackSpy).not.toHaveBeenCalled();
scope.end();
});
it("fires CallReconnecting with homeserver reason and duration when reconnected", async () => {
const scope = new ObservableScope();
const trackSpy = vi.spyOn(
PosthogAnalytics.instance.eventCallReconnecting,
"track",
);
const hsReason$ = new BehaviorSubject<
[boolean, HomeserverDisconnectReason | null]
>([true, null]);
const connectionManagerData = new ConnectionManagerData();
connectionManagerData.add(connectionTransportAConnected, []);
createLocalMembership$({
scope,
...defaultCreateLocalMemberValues,
homeserverConnected: {
combined$: hsReason$,
rtsSession$: constant(RTCMemberStatus.Connected),
},
connectionManager: {
connectionManagerData$: constant(new Epoch(connectionManagerData)),
},
localTransport$: new BehaviorSubject({
advertised$: new BehaviorSubject(aTransport),
active$: new BehaviorSubject(aTransportWithSFUConfig),
}),
});
await flushPromises();
hsReason$.next([false, "sync"]);
hsReason$.next([true, null]);
expect(trackSpy).toHaveBeenCalledWith(
defaultCreateLocalMemberValues.roomId,
"sync",
expect.any(Number),
);
scope.end();
});
it("reports livekit reason when livekit disconnects then reconnects", async () => {
const scope = new ObservableScope();
const trackSpy = vi.spyOn(
PosthogAnalytics.instance.eventCallReconnecting,
"track",
);
const connectionState$ = new BehaviorSubject<ConnectionState>(
ConnectionState.LivekitConnected,
);
const mutableConnection = {
...connectionTransportAConnected,
state$: connectionState$,
} as unknown as Connection;
const connectionManagerData = new ConnectionManagerData();
connectionManagerData.add(mutableConnection, []);
createLocalMembership$({
scope,
...defaultCreateLocalMemberValues,
homeserverConnected: {
combined$: new BehaviorSubject<
[boolean, HomeserverDisconnectReason | null]
>([true, null]),
rtsSession$: constant(RTCMemberStatus.Connected),
},
connectionManager: {
connectionManagerData$: constant(new Epoch(connectionManagerData)),
},
localTransport$: new BehaviorSubject({
advertised$: new BehaviorSubject(aTransport),
active$: new BehaviorSubject(aTransportWithSFUConfig),
}),
});
await flushPromises();
connectionState$.next(ConnectionState.LivekitDisconnected);
connectionState$.next(ConnectionState.LivekitConnected);
expect(trackSpy).toHaveBeenCalledWith(
defaultCreateLocalMemberValues.roomId,
"livekit",
expect.any(Number),
);
scope.end();
});
it("fires one event per completed reconnection cycle", async () => {
const scope = new ObservableScope();
const trackSpy = vi.spyOn(
PosthogAnalytics.instance.eventCallReconnecting,
"track",
);
const hsReason$ = new BehaviorSubject<
[boolean, HomeserverDisconnectReason | null]
>([true, null]);
const connectionManagerData = new ConnectionManagerData();
connectionManagerData.add(connectionTransportAConnected, []);
createLocalMembership$({
scope,
...defaultCreateLocalMemberValues,
homeserverConnected: {
combined$: hsReason$,
rtsSession$: constant(RTCMemberStatus.Connected),
},
connectionManager: {
connectionManagerData$: constant(new Epoch(connectionManagerData)),
},
localTransport$: new BehaviorSubject({
advertised$: new BehaviorSubject(aTransport),
active$: new BehaviorSubject(aTransportWithSFUConfig),
}),
});
await flushPromises();
hsReason$.next([false, "membership"]);
hsReason$.next([true, null]);
hsReason$.next([false, "probablyLeft"]);
hsReason$.next([false, "sync"]);
hsReason$.next([false, "membership"]);
hsReason$.next([true, null]);
expect(trackSpy).toHaveBeenCalledTimes(2);
expect(trackSpy).toHaveBeenNthCalledWith(
1,
defaultCreateLocalMemberValues.roomId,
"membership",
expect.any(Number),
);
expect(trackSpy).toHaveBeenNthCalledWith(
2,
defaultCreateLocalMemberValues.roomId,
"probablyLeft",
expect.any(Number),
);
scope.end();
});
});
});

View File

@@ -55,7 +55,6 @@ import { ElementWidgetActions, widget } from "../../../widget.ts";
import { getUrlParams } from "../../../UrlParams.ts";
import { PosthogAnalytics } from "../../../analytics/PosthogAnalytics.ts";
import {
MatrixRTCMode,
advancedScreenShare,
screenShareResolution,
screenShareFramerate,
@@ -63,6 +62,7 @@ import {
screenShareCodec,
parseResolution,
} from "../../../settings/settings.ts";
import { MatrixRTCMode } from "../../../config/ConfigOptions.ts";
import { Config } from "../../../config/Config.ts";
import {
ConnectionState,
@@ -70,7 +70,6 @@ import {
type FailedToStartError,
} from "../remoteMembers/Connection.ts";
import { type HomeserverConnected } from "./HomeserverConnected.ts";
import { and$ } from "../../../utils/observable.ts";
import { type LocalTransport } from "./LocalTransport.ts";
import { areLivekitTransportsEqual } from "../remoteMembers/MatrixLivekitMembers.ts";
@@ -138,6 +137,7 @@ interface Props {
createPublisherFactory: (connection: Connection) => Publisher;
joinMatrixRTC: (transport: LivekitTransportConfig) => void;
homeserverConnected: HomeserverConnected;
roomId: string;
localTransport$: Behavior<LocalTransport>;
matrixRTCSession: Pick<
MatrixRTCSession,
@@ -161,6 +161,7 @@ interface Props {
* @param props.logger The logger to use.
* @param props.muteStates The mute states for video and audio.
* @param props.matrixRTCSession The matrix RTC session to join.
* @param props.roomId The room ID used as the call identifier in analytics events.
* @returns
* - publisher: The handle to create tracks and publish them to the room.
* - connected$: the current connection state. Including matrix server and livekit server connection. (only considering the livekit server we are using for our own media publication)
@@ -178,6 +179,7 @@ export const createLocalMembership$ = ({
logger: parentLogger,
muteStates,
matrixRTCSession,
roomId: roomId,
}: Props): {
/**
* This request to start audio and video tracks.
@@ -503,20 +505,35 @@ export const createLocalMembership$ = ({
);
/**
* Whether we are "fully" connected to the call. Accounts for both the
* connection to the MatrixRTC session and the LiveKit publish connection.
* The disconnect reason for the combined Matrix + LiveKit connection, or null
* when fully connected. Homeserver reasons take priority over livekit.
* Both connectivity state and reason come from the same combineLatest emission,
* avoiding any race between the two.
*/
const matrixAndLivekitConnected$ = scope.behavior(
and$(
const connectionDisconnectReason$ = scope.behavior(
combineLatest([
homeserverConnected.combined$,
localConnectionState$.pipe(
map((state) => state === ConnectionState.LivekitConnected),
),
).pipe(
]).pipe(
map(([[hsConnected, hsReason], livekitConnected]) => {
if (!hsConnected) return hsReason!;
if (!livekitConnected) return "livekit" as const;
return null;
}),
tap((v) => logger.debug("livekit+matrix: Connected state changed", v)),
),
);
/**
* Whether we are "fully" connected to the call. Accounts for both the
* connection to the MatrixRTC session and the LiveKit publish connection.
*/
const matrixAndLivekitConnected$ = scope.behavior(
connectionDisconnectReason$.pipe(map((reason) => reason === null)),
);
/**
* Whether we should tell the user that we're reconnecting to the call.
*/
@@ -528,6 +545,33 @@ export const createLocalMembership$ = ({
false,
);
let reconnectStart: {
time: number;
reason: NonNullable<(typeof connectionDisconnectReason$)["value"]>;
} | null = null;
connectionDisconnectReason$
.pipe(distinctUntilChanged(), pairwise(), scope.bind())
.subscribe(([prev, reason]) => {
if (reason !== null) {
// Only begin tracking when transitioning FROM connected (null → non-null).
// This prevents the initial startup phase — where we may be non-null before
// the first real connection — from being counted as a reconnect.
if (prev === null) {
reconnectStart ??= { time: Date.now(), reason };
}
} else if (reconnectStart !== null) {
PosthogAnalytics.instance.eventCallReconnecting.track(
roomId,
reconnectStart.reason,
(Date.now() - reconnectStart.time) / 1000,
);
PosthogAnalytics.instance.eventCallEnded.cacheReconnecting(
reconnectStart.reason,
);
reconnectStart = null;
}
});
// inform the widget about the connect and disconnect intent from the user.
scope
.behavior(joinAndPublishRequested$.pipe(pairwise(), scope.bind()), [
@@ -615,7 +659,7 @@ export const createLocalMembership$ = ({
// TODO refactor this based no livekitState$
combineLatest([participant$, homeserverConnected.combined$])
.pipe(scope.bind())
.subscribe(([participant, connected]) => {
.subscribe(([participant, [connected]]) => {
if (!participant) return;
const publications = participant.trackPublications.values();
if (connected) {
@@ -828,6 +872,19 @@ export function enterRTCSession(
};
}
// Calculates `maximumNetworkErrorRetryCount`. The connection is failed if EITHER:
// - The /sync loop is unresponsive for > `gracePeriod` ms, or
// - A delayed leave event is emitted (after `leaveDelay` ms period).
// Note: Use leaveDelay >> gracePeriod for delegated leave events.
const gracePeriod = Config.get().sync_disconnect_grace_period_ms;
const leaveDelay = matrixRtcSessionConfig?.delayed_leave_event_delay_ms;
const retryInterval = matrixRtcSessionConfig?.network_error_retry_ms;
// Math.min is used to account for the respective worst case: /sync not available or leave event emitted.
const maxWaitTime = Math.min(gracePeriod, leaveDelay);
const maximumNetworkErrorRetryCount =
Math.ceil(maxWaitTime / retryInterval) + 1;
// Multi-sfu does not need a preferred foci list. just the focus that is actually used.
// TODO where/how do we track errors originating from the ongoing rtcSession?
@@ -853,6 +910,7 @@ export function enterRTCSession(
membershipEventExpiryMs:
matrixRtcSessionConfig?.membership_event_expiry_ms,
unstableSendStickyEvents: matrixRTCMode === MatrixRTCMode.Matrix_2_0,
maximumNetworkErrorRetryCount: maximumNetworkErrorRetryCount,
},
);
}

View File

@@ -63,6 +63,7 @@ function createMockLocalTrack(source: Track.Source): LocalTrack {
function createMockMuteState(enabled$: BehaviorSubject<boolean>): {
enabled$: BehaviorSubject<boolean>;
syncing$: BehaviorSubject<boolean>;
setHandler: (h: (enabled: boolean) => void) => void;
unsetHandler: () => void;
} {
@@ -70,6 +71,7 @@ function createMockMuteState(enabled$: BehaviorSubject<boolean>): {
const ms = {
enabled$,
syncing$: new BehaviorSubject(false),
setHandler: vi.fn().mockImplementation((h: (enabled: boolean) => void) => {
currentHandler = h;
}),
@@ -132,6 +134,7 @@ beforeEach(() => {
new Map(),
{},
{},
{},
);
vi.mocked(localParticipant).createTracks = vi

View File

@@ -112,27 +112,49 @@ export class Publisher {
this.logger.info("Local track published", localTrackPublication);
const lkRoom = this.connection.livekitRoom;
if (!this.shouldPublish) {
this.logger.debug("Not publishing, pausing upstream");
this.pauseUpstreams(lkRoom, [localTrackPublication.source]).catch((e) => {
this.logger.error(`Failed to pause upstreams`, e);
});
}
// also check the mute state and apply it
if (localTrackPublication.source === Track.Source.Microphone) {
const enabled = this.muteStates.audio.enabled$.value;
lkRoom.localParticipant.setMicrophoneEnabled(enabled).catch((e) => {
this.logger.error(
`Failed to enable microphone track, enabled:${enabled}`,
e,
);
});
const muteState = this.muteStates.audio;
// skip this if a sync is in progress: enabled$ still reflects the old
// state while the handler is mid-flight, so the handler itself will apply
// the correct mute state once it completes.
if (!muteState.syncing$.value) {
const enabled = muteState.enabled$.value;
if (!enabled) {
this.logger.info(
"Local audio track just published but muted meanwhile, setting enabled to false",
);
lkRoom.localParticipant.setMicrophoneEnabled(false).catch((e) => {
this.logger.error(
`Failed to enable microphone track, enabled:${enabled}`,
e,
);
});
}
}
} else if (localTrackPublication.source === Track.Source.Camera) {
const enabled = this.muteStates.video.enabled$.value;
lkRoom.localParticipant.setCameraEnabled(enabled).catch((e) => {
this.logger.error(
`Failed to enable camera track, enabled:${enabled}`,
e,
);
});
const muteState = this.muteStates.video;
// skip this if a sync is in progress: enabled$ still reflects the old
// state while the handler is mid-flight, so the handler itself will apply
// the correct mute state once it completes.
if (!muteState.syncing$.value) {
const enabled = muteState.enabled$.value;
if (!enabled) {
this.logger.info(
"Local video track just published but muted meanwhile, setting enabled to false",
);
lkRoom.localParticipant.setCameraEnabled(false).catch((e) => {
this.logger.error(
`Failed to enable camera track, enabled:${enabled}`,
e,
);
});
}
}
}
}
/**
@@ -379,10 +401,11 @@ export class Publisher {
if (!this.shouldPublish && enable) {
await this.pauseUpstreams(lkRoom, [Track.Source.Microphone]);
}
return enable;
} catch (e) {
this.logger.error("Failed to update LiveKit audio input mute state", e);
return lkRoom.localParticipant.isMicrophoneEnabled;
}
return lkRoom.localParticipant.isMicrophoneEnabled;
});
this.muteStates.video.setHandler(async (enable) => {
try {
@@ -393,10 +416,11 @@ export class Publisher {
if (!this.shouldPublish && enable) {
await this.pauseUpstreams(lkRoom, [Track.Source.Camera]);
}
return enable;
} catch (e) {
this.logger.error("Failed to update LiveKit video input mute state", e);
return lkRoom.localParticipant.isCameraEnabled;
}
return lkRoom.localParticipant.isCameraEnabled;
});
}