[SDK] update state structure (less duplication, one state enum)

- remove media from LocalMembership file
This commit is contained in:
Timo K.
2026-10-08 13:53:57 +02:00
parent f49c1cd3d4
commit 3cdb452d5d
12 changed files with 500 additions and 497 deletions
+29 -21
View File
@@ -14,15 +14,17 @@ import {
} from "matrix-js-sdk/lib/matrixrtc";
import { type IMembershipManager } from "matrix-js-sdk/lib/matrixrtc/IMembershipManager";
import { type CallMembershipIdentityParts } from "matrix-js-sdk/lib/matrixrtc/EncryptionManager";
import { deepCompare } from "matrix-js-sdk/lib/utils";
import { v4 as uuidv4 } from "uuid";
import {
BehaviorSubject,
combineLatest,
distinctUntilChanged,
filter,
from,
fromEvent,
map,
Observable,
scan,
} from "rxjs";
import { defaultSessionTimings, MatrixRTCMode } from "./config";
@@ -34,6 +36,7 @@ import { mapScoped } from "./utils/mapScoped";
import {
type DataMessage,
type LocalRTCMember,
type ParticipationState,
type RemoteRTCMember,
type RTCMember,
type RTCParticipation,
@@ -59,8 +62,8 @@ import {
createRemoteRTCMember,
} from "./matrixrtc/Members";
import { createKeyRotationSuppressed$ } from "./matrixrtc/JsRtcSessionBehaviors";
import { fatalError, participationStatus } from "./matrixrtc/status";
import { type TransportRegistry } from "./matrixrtc/Transports";
import { participationState } from "./participationState";
/** What the slot lends a participation. */
export interface ParticipationDeps {
@@ -124,7 +127,7 @@ export function createRTCParticipation(
mediaKeys$: mediaKeys$(jsSdkSession),
timings,
logger,
});
});
const preparedTransport$ = from(
discoverLocalTransport(client, backend.transportType, options, logger).then(
@@ -137,7 +140,6 @@ export function createRTCParticipation(
const localMembership = createLocalMembership$({
scope,
local: backend.local,
delegateDelayedLeave: async (delayId) =>
backend.delegateDelayedLeave(delayId),
preparedTransport$,
@@ -228,22 +230,32 @@ export function createRTCParticipation(
filter((message) => message !== null),
);
const left$ = new BehaviorSubject(false);
const status$ = scope.behavior(
combineLatest(
[
localMembership.state$,
localMembership.connected$,
localMembership.reconnecting$,
left$,
],
participationStatus,
// Nothing leaves this device while it may already have been dropped from
// the session: the member would show as away while still being heard
localMembership.state$
.pipe(
map((membership) => membership.kind === "joined"),
distinctUntilChanged(),
scope.bind(),
)
.subscribe((publish) => backend.local.setPublishing(publish));
const state$ = scope.behavior<ParticipationState>(
combineLatest([
localMembership.state$,
backend.local.connectionState$,
]).pipe(
scan(
(previous, [membership, media]) =>
participationState(previous, membership, media),
{ kind: "waitingForTransport" } as ParticipationState,
),
distinctUntilChanged(deepCompare),
),
);
const leave = (): void => {
if (left$.value) return;
left$.next(true);
if (state$.value.kind === "left") return;
localMembership.leave();
scope.end();
};
@@ -253,11 +265,7 @@ export function createRTCParticipation(
participation: {
slot,
leave,
status$,
connected$: localMembership.connected$,
reconnecting$: localMembership.reconnecting$,
disconnectReason$: localMembership.disconnectReason$,
fatalError$: scope.behavior(localMembership.state$.pipe(map(fatalError))),
state$,
localMember$,
remoteMembers$,
publish: async (request) => backend.local.publish(request),
+31 -18
View File
@@ -121,31 +121,44 @@ export interface RTCParticipationOptions {
backend?: MediaBackendFactory;
}
export type ParticipationStatus =
| "waitingForTransport"
| "connecting"
| "connected"
| "reconnecting"
| "disconnected"
/** `leave()` was called; the participation is over and stays so. */
| "left";
/**
* The homeserver link that is down, in the order they depend on each other:
* without sync the membership is unknowable, and without a confirmed
* membership the delayed leave cannot be refreshed.
*/
export type MatrixDisconnectReason = "sync" | "membership" | "probablyLeft";
/** Why the participation is not connected: the first failing of its three links. */
export type DisconnectReason = "sync" | "membership" | "probablyLeft" | "media";
/**
* Why the participation is not connected: the first failing link, in the
* order they depend on each other. The media connection comes last because
* it is worth little while the Matrix side is down.
*/
export type DisconnectReason = MatrixDisconnectReason | "media";
export type ParticipationState =
/** Asking the homeserver for a transport, or preparing it. */
| { kind: "waitingForTransport" }
/** Transport known; joining the session and connecting the media for the first time. */
| { kind: "connecting" }
/** Membership joined and the local media connection up. */
| { kind: "connected" }
/** Was connected and one link is down. Back to `connected` or on to `failed`. */
| { kind: "reconnecting"; reason: DisconnectReason }
/**
* The membership or the media connection ended with an error. Terminal
* unless `left`. A failed publication is not fatal: the member can still
* receive, so it stays on the local member's media.
*/
| { kind: "failed"; error: MatrixRTCError }
/** `leave()` was called. Terminal. */
| { kind: "left" };
/** The span between our `join()` of a slot and our `leave()`. */
export interface RTCParticipation {
readonly slot: RTCSlot;
/** Sends the leave, tears the connections down and ends the participation. */
leave(): void;
/** Collapsed view of the local member's state machine. */
status$: Behavior<ParticipationStatus>;
connected$: Behavior<boolean>;
reconnecting$: Behavior<boolean>;
/** Null while connected, the first failing link otherwise. */
disconnectReason$: Behavior<DisconnectReason | null>;
/** A transport, Matrix or connection error that stops the participation. */
fatalError$: Behavior<MatrixRTCError | null>;
state$: Behavior<ParticipationState>;
/** Null until our own membership has been seen in the room. */
localMember$: Behavior<LocalRTCMember | null>;
+74 -41
View File
@@ -100,13 +100,13 @@ describe("createHomeserverConnected$", () => {
// 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"]);
expect(hsConnected.disconnectReason$.value).toEqual("sync");
});
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).toEqual([false, "membership"]);
expect(hsConnected.disconnectReason$.value).toEqual("membership");
});
it("reports probablyLeft reason when membership transitions to Connected but ProbablyLeft is true", () => {
@@ -116,17 +116,17 @@ describe("createHomeserverConnected$", () => {
// Indicate probable leave before connection
session.setProbablyLeft(true);
session.setMembershipStatus(Status.Connected);
expect(hsConnected.combined$.value).toEqual([false, "probablyLeft"]);
expect(hsConnected.disconnectReason$.value).toEqual("probablyLeft");
});
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).toEqual([false, "membership"]); // not yet membership connected
expect(hsConnected.disconnectReason$.value).toEqual("membership"); // not yet membership connected
// 2. Membership connected
session.setMembershipStatus(Status.Connected);
expect(hsConnected.combined$.value).toEqual([true, null]); // probablyLeft is false
expect(hsConnected.disconnectReason$.value).toEqual(null); // probablyLeft is false
});
it("returns syncing reason when sync loop leaves Syncing", () => {
@@ -134,76 +134,76 @@ describe("createHomeserverConnected$", () => {
// Reach connected state
client.setSyncState(SyncState.Syncing);
session.setMembershipStatus(Status.Connected);
expect(hsConnected.combined$.value).toEqual([true, null]);
expect(hsConnected.disconnectReason$.value).toEqual(null);
// Sync loop error => should report syncing reason
client.setSyncState(SyncState.Error);
expect(hsConnected.combined$.value).toEqual([false, "sync"]);
expect(hsConnected.disconnectReason$.value).toEqual("sync");
});
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).toEqual([true, null]);
expect(hsConnected.disconnectReason$.value).toEqual(null);
session.setMembershipStatus(Status.Disconnected);
expect(hsConnected.combined$.value).toEqual([false, "membership"]);
expect(hsConnected.disconnectReason$.value).toEqual("membership");
});
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).toEqual([true, null]);
expect(hsConnected.disconnectReason$.value).toEqual(null);
session.setProbablyLeft(true);
expect(hsConnected.combined$.value).toEqual([false, "probablyLeft"]);
expect(hsConnected.disconnectReason$.value).toEqual("probablyLeft");
});
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).toEqual([true, null]);
expect(hsConnected.disconnectReason$.value).toEqual(null);
session.setProbablyLeft(true);
expect(hsConnected.combined$.value).toEqual([false, "probablyLeft"]);
expect(hsConnected.disconnectReason$.value).toEqual("probablyLeft");
// Simulate clearing the flag (in realistic scenario membership manager would update)
session.setProbablyLeft(false);
expect(hsConnected.combined$.value).toEqual([true, null]);
expect(hsConnected.disconnectReason$.value).toEqual(null);
});
it("composite sequence reflects each individual failure reason", () => {
const hsConnected = createHomeserverConnected$(scope, client, session, 0);
// Initially: sync error + membership disconnected → syncing wins (highest priority)
expect(hsConnected.combined$.value).toEqual([false, "sync"]);
expect(hsConnected.disconnectReason$.value).toEqual("sync");
// Fix sync only → membershipConnected is now the blocker
client.setSyncState(SyncState.Syncing);
expect(hsConnected.combined$.value).toEqual([false, "membership"]);
expect(hsConnected.disconnectReason$.value).toEqual("membership");
// Fix membership → all conditions satisfied
session.setMembershipStatus(Status.Connected);
expect(hsConnected.combined$.value).toEqual([true, null]);
expect(hsConnected.disconnectReason$.value).toEqual(null);
// Introduce probablyLeft → certainlyConnected
session.setProbablyLeft(true);
expect(hsConnected.combined$.value).toEqual([false, "probablyLeft"]);
expect(hsConnected.disconnectReason$.value).toEqual("probablyLeft");
// Restore notProbablyLeft → connected again
session.setProbablyLeft(false);
expect(hsConnected.combined$.value).toEqual([true, null]);
expect(hsConnected.disconnectReason$.value).toEqual(null);
// Drop sync → syncing reason
client.setSyncState(SyncState.Error);
expect(hsConnected.combined$.value).toEqual([false, "sync"]);
expect(hsConnected.disconnectReason$.value).toEqual("sync");
});
});
describe("createHomeserverConnected$ - combined$ reason values", () => {
describe("createHomeserverConnected$ - disconnectReason$ reason values", () => {
let scope: ObservableScope;
let client: MockMatrixClient;
let session: MockMatrixRTCSession;
@@ -222,57 +222,87 @@ describe("createHomeserverConnected$ - combined$ reason values", () => {
scope.end();
});
it("is [true, null] when all three conditions are satisfied", () => {
const { combined$ } = createHomeserverConnected$(scope, client, session, 0);
it("is null when all three conditions are satisfied", () => {
const { disconnectReason$ } = createHomeserverConnected$(
scope,
client,
session,
0,
);
client.setSyncState(SyncState.Syncing);
session.setMembershipStatus(Status.Connected);
expect(combined$.value).toEqual([true, null]);
expect(disconnectReason$.value).toEqual(null);
});
it("reports syncing when sync loop is not Syncing", () => {
const { combined$ } = createHomeserverConnected$(scope, client, session, 0);
const { disconnectReason$ } = createHomeserverConnected$(
scope,
client,
session,
0,
);
// client starts with SyncState.Error, membership also disconnected
expect(combined$.value).toEqual([false, "sync"]);
expect(disconnectReason$.value).toEqual("sync");
});
it("reports membershipConnected when sync is fine but membership is not Connected", () => {
const { combined$ } = createHomeserverConnected$(scope, client, session, 0);
const { disconnectReason$ } = createHomeserverConnected$(
scope,
client,
session,
0,
);
client.setSyncState(SyncState.Syncing);
// session still Status.Disconnected
expect(combined$.value).toEqual([false, "membership"]);
expect(disconnectReason$.value).toEqual("membership");
});
it("reports certainlyConnected when probablyLeft is true", () => {
const { combined$ } = createHomeserverConnected$(scope, client, session, 0);
const { disconnectReason$ } = createHomeserverConnected$(
scope,
client,
session,
0,
);
client.setSyncState(SyncState.Syncing);
session.setMembershipStatus(Status.Connected);
session.setProbablyLeft(true);
expect(combined$.value).toEqual([false, "probablyLeft"]);
expect(disconnectReason$.value).toEqual("probablyLeft");
});
it("prioritises syncing over membershipConnected when both fail", () => {
const { combined$ } = createHomeserverConnected$(scope, client, session, 0);
const { disconnectReason$ } = createHomeserverConnected$(
scope,
client,
session,
0,
);
// Both sync (Error) and membership (Disconnected) are failing
expect(combined$.value).toEqual([false, "sync"]);
expect(disconnectReason$.value).toEqual("sync");
});
it("updates reason as conditions change", () => {
const { combined$ } = createHomeserverConnected$(scope, client, session, 0);
const { disconnectReason$ } = createHomeserverConnected$(
scope,
client,
session,
0,
);
// Initially: syncing fails
expect(combined$.value).toEqual([false, "sync"]);
expect(disconnectReason$.value).toEqual("sync");
// Fix sync → membershipConnected is now the blocker
client.setSyncState(SyncState.Syncing);
expect(combined$.value).toEqual([false, "membership"]);
expect(disconnectReason$.value).toEqual("membership");
// Fix membership → probablyLeft makes certainlyConnected fail
session.setProbablyLeft(true);
session.setMembershipStatus(Status.Connected);
expect(combined$.value).toEqual([false, "probablyLeft"]);
expect(disconnectReason$.value).toEqual("probablyLeft");
// Clear probablyLeft → all conditions satisfied
session.setProbablyLeft(false);
expect(combined$.value).toEqual([true, null]);
expect(disconnectReason$.value).toEqual(null);
});
});
@@ -303,10 +333,13 @@ describe("createHomeserverConnected$ - Grace Period", () => {
session,
GRACE_PERIOD,
);
expectObservable(hsConnected.combined$).toBe(expectedConnectedMarbles, {
y: [true, null],
n: [false, "sync"],
});
expectObservable(hsConnected.disconnectReason$).toBe(
expectedConnectedMarbles,
{
y: null,
n: "sync",
},
);
});
}
+14 -26
View File
@@ -29,6 +29,7 @@ import { logger as rootLogger } from "matrix-js-sdk/lib/logger";
import { type ObservableScope } from "../reactive/ObservableScope";
import { type Behavior } from "../reactive/Behavior";
import { defaultSessionTimings } from "../config";
import { type MatrixDisconnectReason } from "../api";
type NodeEventHandler = (...args: unknown[]) => void;
@@ -37,22 +38,16 @@ interface NodeStyleEventEmitter {
removeListener(eventName: string | symbol, handler: NodeEventHandler): this;
}
export type HomeserverDisconnectReason = "sync" | "membership" | "probablyLeft";
export interface HomeserverConnected {
/**
* 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>;
/** Null while the homeserver connection is healthy, the first failing link otherwise. */
disconnectReason$: Behavior<MatrixDisconnectReason | null>;
}
/**
* Behavior representing whether we consider ourselves connected to the Matrix homeserver
* for the purposes of a MatrixRTC session.
*
* `combined$` emits `null` when all conditions are satisfied, or the first failing
* `disconnectReason$` 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"
@@ -118,26 +113,19 @@ export function createHomeserverConnected$(
map(() => matrixRTCSession.probablyLeft !== true),
);
const combined$ = scope.behavior(
const disconnectReason$ = scope.behavior(
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}`,
);
map(([syncing, membership, certainly]): MatrixDisconnectReason | null => {
if (!syncing) return "sync";
if (!membership) return "membership";
if (!certainly) return "probablyLeft";
return null;
}),
tap((reason) => {
logger.info(`Homeserver connected update: ${reason ?? "connected"}`);
}),
),
);
return { combined$, rtsSession$ };
return { disconnectReason$ };
}
+56 -102
View File
@@ -5,77 +5,57 @@ SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
Please see LICENSE in the repository root for full details.
*/
import { logger } from "matrix-js-sdk/lib/logger";
import { Status, type Transport } from "matrix-js-sdk/lib/matrixrtc";
import {
BehaviorSubject,
EMPTY,
type Observable,
Subject,
throwError,
} from "rxjs";
import { describe, expect, it, vi } from "vitest";
import { type Transport } from "matrix-js-sdk/lib/matrixrtc";
import { logger } from "matrix-js-sdk/lib/logger";
import { BehaviorSubject, type Observable, Subject, throwError } from "rxjs";
import { defaultSessionTimings } from "../config";
import { MatrixRTCError } from "../errors";
import {
type LocalMediaBackend,
MediaConnectionState,
} from "../media-backend/api";
import { type HomeserverConnected } from "./HomeserverConnected";
import {
createLocalMembership$,
type LocalMembership,
type PreparedTransport,
PublishState,
TransportState,
} from "./LocalMembership";
import { flushPromises, testScope } from "../utils/test";
const transport: Transport = { type: "livekit", livekit_service_url: "u" };
describe("createLocalMembership$", () => {
it("waits for the transport, then reports the media connection and the join", async () => {
const { membership, local, prepared$ } = setup();
expect(membership.state$.value).toBe(TransportState.Waiting);
it("waits for the transport, joins, and leaves on leave()", async () => {
const { membership, homeserver, prepared$ } = setup();
expect(membership.state$.value).toEqual({ kind: "waitingForTransport" });
homeserver.disconnectReason$.next("membership");
prepared$.next({ transport, canDelegateDelayedLeave: false });
expect(membership.state$.value).toEqual({
media: { connection: MediaConnectionState.Initialized },
matrix: Status.Connected,
});
expect(membership.state$.value).toEqual({ kind: "joining" });
local.connectionState$.next(MediaConnectionState.Connected);
await flushPromises();
expect(membership.state$.value).toEqual({
media: PublishState.Publishing,
matrix: Status.Connected,
});
expect(membership.connected$.value).toBe(true);
homeserver.disconnectReason$.next(null);
expect(membership.state$.value).toEqual({ kind: "joined" });
membership.leave();
expect(membership.state$.value).toEqual({
media: PublishState.WaitingForUser,
matrix: Status.Connected,
});
await flushPromises();
expect(membership.state$.value).toEqual({ kind: "left" });
});
it("publishes only while joined and the homeserver is reachable", async () => {
const { membership, local, homeserver } = setup();
expect(local.setPublishing).toHaveBeenLastCalledWith(true);
it("reconnects only after it was joined once", () => {
const { membership, homeserver, prepared$ } = setup();
prepared$.next({ transport, canDelegateDelayedLeave: false });
expect(membership.state$.value).toEqual({ kind: "joined" });
homeserver.combined$.next([false, "sync"]);
expect(local.setPublishing).toHaveBeenLastCalledWith(false);
homeserver.combined$.next([true, null]);
expect(local.setPublishing).toHaveBeenLastCalledWith(true);
membership.leave();
expect(local.setPublishing).toHaveBeenLastCalledWith(false);
// A sync outage after leaving must not bring the media back
homeserver.combined$.next([false, "sync"]);
homeserver.combined$.next([true, null]);
expect(local.setPublishing).toHaveBeenLastCalledWith(false);
await flushPromises();
homeserver.disconnectReason$.next("sync");
expect(membership.state$.value).toEqual({
kind: "reconnecting",
reason: "sync",
});
homeserver.disconnectReason$.next("probablyLeft");
expect(membership.state$.value).toEqual({
kind: "reconnecting",
reason: "probablyLeft",
});
homeserver.disconnectReason$.next(null);
expect(membership.state$.value).toEqual({ kind: "joined" });
});
it("joins with the delegated timings and hands over each delay id where the backend can", async () => {
@@ -107,57 +87,54 @@ describe("createLocalMembership$", () => {
expect(delegate).not.toHaveBeenCalled();
});
it("reports a transport that could not be prepared as fatal", () => {
it("reports a transport that could not be prepared as failed", () => {
const error = new MatrixRTCError("no token");
const { membership } = setup(throwError(() => error));
expect(membership.state$.value).toBe(error);
expect(membership.state$.value).toEqual({ kind: "failed", error });
});
it("reports a connection loss and a publish failure in the media state", () => {
const { membership, local, prepared$ } = setup();
it("reports the membership manager giving up as failed, and stays failed", () => {
const { membership, homeserver, prepared$, membershipManagerError$ } =
setup();
prepared$.next({ transport, canDelegateDelayedLeave: false });
local.connectionState$.next(MediaConnectionState.Connected);
local.connectionState$.next(MediaConnectionState.Reconnecting);
expect(membership.state$.value).toEqual({
media: { connection: MediaConnectionState.Reconnecting },
matrix: Status.Connected,
});
expect(membership.disconnectReason$.value).toBe("media");
expect(membership.reconnecting$.value).toBe(true);
const failure = new MatrixRTCError("no camera");
local.publishError$.next(failure);
membershipManagerError$.next(new Error("gave up"));
const state = membership.state$.value;
expect(typeof state === "object" && "media" in state && state.media).toBe(
failure,
);
expect(state.kind).toBe("failed");
homeserver.disconnectReason$.next("sync");
homeserver.disconnectReason$.next(null);
expect(membership.state$.value).toBe(state);
});
it("is left after a failure once leave() is called", () => {
const { membership } = setup(throwError(() => new MatrixRTCError("x")));
membership.leave();
expect(membership.state$.value).toEqual({ kind: "left" });
});
});
function setup(preparedTransportWithErrors$?: Observable<PreparedTransport>): {
membership: LocalMembership;
local: FakeLocalMediaBackend;
homeserver: FakeHomeserverConnected;
prepared$: Subject<PreparedTransport>;
delayId$: BehaviorSubject<string | null>;
membershipManagerError$: Subject<unknown>;
joinMatrixRTC: ReturnType<typeof vi.fn>;
delegate: ReturnType<typeof vi.fn>;
} {
const prepared$ = new Subject<PreparedTransport>();
const local = fakeLocalMediaBackend();
const homeserver = fakeHomeserverConnected();
const delayId$ = new BehaviorSubject<string | null>(null);
const membershipManagerError$ = new Subject<unknown>();
const joinMatrixRTC = vi.fn();
const delegate = vi.fn(async () => Promise.resolve());
const membership = createLocalMembership$({
scope: testScope(),
local,
delegateDelayedLeave: delegate,
preparedTransport$: preparedTransportWithErrors$ ?? prepared$,
homeserverConnected: homeserver,
joinMatrixRTC,
membershipManagerError$: new Subject(),
membershipManagerError$,
matrixRTCSession: {
leaveRoomSession: vi.fn(async () => Promise.resolve(true)),
},
@@ -167,48 +144,25 @@ function setup(preparedTransportWithErrors$?: Observable<PreparedTransport>): {
});
return {
membership,
local,
homeserver,
prepared$,
delayId$,
membershipManagerError$,
joinMatrixRTC,
delegate,
};
}
type FakeLocalMediaBackend = Omit<
LocalMediaBackend,
"connectionState$" | "publishError$" | "setPublishing"
> & {
connectionState$: BehaviorSubject<MediaConnectionState | Error>;
publishError$: BehaviorSubject<Error | null>;
setPublishing: ReturnType<typeof vi.fn<LocalMediaBackend["setPublishing"]>>;
};
function fakeLocalMediaBackend(): FakeLocalMediaBackend {
return {
connectionState$: new BehaviorSubject<MediaConnectionState | Error>(
MediaConnectionState.Initialized,
),
tracks$: new BehaviorSubject(null),
encryptionError$: EMPTY,
setPublishing: vi.fn<LocalMediaBackend["setPublishing"]>(),
publishError$: new BehaviorSubject<Error | null>(null),
publish: vi.fn<LocalMediaBackend["publish"]>(),
unpublish: vi.fn<LocalMediaBackend["unpublish"]>(),
};
}
type FakeHomeserverConnected = Omit<HomeserverConnected, "combined$"> & {
combined$: BehaviorSubject<HomeserverConnected["combined$"]["value"]>;
type FakeHomeserverConnected = {
disconnectReason$: BehaviorSubject<
HomeserverConnected["disconnectReason$"]["value"]
>;
};
function fakeHomeserverConnected(): FakeHomeserverConnected {
return {
combined$: new BehaviorSubject<HomeserverConnected["combined$"]["value"]>([
true,
null,
]),
rtsSession$: new BehaviorSubject(Status.Connected),
disconnectReason$: new BehaviorSubject<
HomeserverConnected["disconnectReason$"]["value"]
>(null),
};
}
+67 -123
View File
@@ -8,7 +8,6 @@ Please see LICENSE in the repository root for full details.
import { type Logger } from "matrix-js-sdk/lib/logger";
import {
type MatrixRTCSession as JsSdkRTCSession,
type Status as RTCSessionStatus,
type Transport,
} from "matrix-js-sdk/lib/matrixrtc";
import { deepCompare } from "matrix-js-sdk/lib/utils";
@@ -16,53 +15,40 @@ import {
BehaviorSubject,
catchError,
combineLatest,
concat,
distinctUntilChanged,
map,
NEVER,
type Observable,
of,
pairwise,
race,
Subject,
switchMap,
scan,
} from "rxjs";
import { type Behavior } from "../reactive/Behavior";
import { type ObservableScope } from "../reactive/ObservableScope";
import { type DelayedLeaveTimings, type SessionTimings } from "../config";
import { type DisconnectReason } from "../api";
import { type MatrixDisconnectReason } from "../api";
import {
type MatrixRTCError,
MembershipManagerError,
toMatrixRTCError,
} from "../errors";
import {
type LocalMediaBackend,
type MediaBackend,
MediaConnectionState,
type TransportCapabilities,
} from "../media-backend/api";
import { type HomeserverConnected } from "./HomeserverConnected";
export enum TransportState {
Waiting = "transport_waiting",
}
export enum PublishState {
WaitingForUser = "publish_waiting_for_user",
Publishing = "publish_publishing",
}
export type LocalMemberMediaState =
| { connection: MediaConnectionState | MatrixRTCError }
| PublishState
| MatrixRTCError;
export type LocalMemberState =
| MatrixRTCError
| TransportState.Waiting
| { media: LocalMemberMediaState; matrix: MatrixRTCError | RTCSessionStatus };
export type MembershipState =
/** Asking the homeserver for a transport, or preparing it. */
| { kind: "waitingForTransport" }
/** Transport known; sending the membership for the first time. */
| { kind: "joining" }
/** Syncing, membership confirmed by the server, delayed leave still being refreshed. */
| { kind: "joined" }
/** Was joined and one homeserver link is down; the membership manager is retrying. */
| { kind: "reconnecting"; reason: MatrixDisconnectReason }
/** A transport, join or membership manager error ended the membership. Terminal unless `left`. */
| { kind: "failed"; error: MatrixRTCError }
/** `leave()` was called. Terminal. */
| { kind: "left" };
/** The local transport, found and prepared, with what the join needs to know about it. */
export interface PreparedTransport extends TransportCapabilities {
@@ -71,7 +57,6 @@ export interface PreparedTransport extends TransportCapabilities {
interface Props {
scope: ObservableScope;
local: LocalMediaBackend;
delegateDelayedLeave: MediaBackend["delegateDelayedLeave"];
preparedTransport$: Observable<PreparedTransport>;
homeserverConnected: HomeserverConnected;
@@ -89,23 +74,18 @@ interface Props {
}
export interface LocalMembership {
/** Sends the leave and stops publishing. Final: there is no joining again. */
/** Sends the leave. Final: there is no joining again. */
leave: () => void;
state$: Behavior<LocalMemberState>;
/** Fully connected: to the homeserver, the session and the transport. */
connected$: Behavior<boolean>;
/** Connected once, and currently not. */
reconnecting$: Behavior<boolean>;
disconnectReason$: Behavior<DisconnectReason | null>;
state$: Behavior<MembershipState>;
}
/**
* The local member's state machine: waits for the transport, then publishes
* and enters the MatrixRTC session, and leaves both in step on `leave()`.
* The Matrix side of the local member: waits for the transport, enters the
* MatrixRTC session, keeps the delayed leave delegated, and leaves on
* `leave()`. Knows nothing about the media.
*/
export function createLocalMembership$({
scope,
local,
delegateDelayedLeave,
preparedTransport$: preparedTransportWithErrors$,
homeserverConnected,
@@ -118,30 +98,20 @@ export function createLocalMembership$({
}: Props): LocalMembership {
const logger = parentLogger.getChild("[LocalMembership]");
const fatalTransportError$ = new Subject<MatrixRTCError>();
const preparedTransport$ = preparedTransportWithErrors$.pipe(
catchError((e: unknown) => {
fatalTransportError$.next(toMatrixRTCError(e));
return NEVER;
}),
);
const transportError$ = new BehaviorSubject<MatrixRTCError | null>(null);
const transport$ = scope.behavior<PreparedTransport | null>(
preparedTransport$,
preparedTransportWithErrors$.pipe(
catchError((e: unknown) => {
transportError$.next(toMatrixRTCError(e));
return NEVER;
}),
),
null,
);
const joined$ = new BehaviorSubject(true);
const matrixError$ = new BehaviorSubject<MatrixRTCError | null>(null);
// Nothing leaves this device while it may already have been dropped from
// the session: the member would show as away while still being heard
combineLatest(
[joined$, homeserverConnected.combined$],
(joined, [connected]) => joined && connected,
)
.pipe(distinctUntilChanged(), scope.bind())
.subscribe((publish) => local.setPublishing(publish));
scope.reconcile(
scope.behavior(combineLatest([transport$, joined$])),
async ([prepared, joined]) => {
@@ -189,78 +159,52 @@ export function createLocalMembership$({
);
});
const mediaState$ = scope.behavior<LocalMemberMediaState>(
combineLatest([local.connectionState$, joined$]).pipe(
map(([connectionState, joined]) => {
if (connectionState !== MediaConnectionState.Connected)
return {
connection:
connectionState instanceof Error
? toMatrixRTCError(connectionState)
: connectionState,
};
return joined ? PublishState.Publishing : PublishState.WaitingForUser;
}),
const state$ = scope.behavior<MembershipState>(
combineLatest([
joined$,
transportError$,
matrixError$,
transport$,
homeserverConnected.disconnectReason$,
]).pipe(
scan(
(previous, [joined, transportError, matrixError, transport, reason]) =>
membershipState(previous, {
joined,
error: transportError ?? matrixError,
hasTransport: transport !== null,
reason,
}),
{ kind: "waitingForTransport" } as MembershipState,
),
distinctUntilChanged(deepCompare),
),
);
const state$ = scope.behavior<LocalMemberState>(
concat(
of(TransportState.Waiting),
race(
fatalTransportError$,
preparedTransport$.pipe(
switchMap(() =>
combineLatest(
[
mediaState$,
homeserverConnected.rtsSession$,
matrixError$,
local.publishError$,
],
(media, sessionStatus, matrixError, publishError) => ({
matrix: matrixError ?? sessionStatus,
media: publishError ? toMatrixRTCError(publishError) : media,
}),
),
),
),
),
),
);
const disconnectReason$ = scope.behavior(
combineLatest(
[homeserverConnected.combined$, local.connectionState$],
(
[homeserverConnected, reason],
connectionState,
): DisconnectReason | null => {
if (!homeserverConnected) return reason ?? "sync";
if (connectionState !== MediaConnectionState.Connected) return "media";
return null;
},
),
);
const connected$ = scope.behavior(
disconnectReason$.pipe(map((reason) => reason === null)),
);
const reconnecting$ = scope.behavior(
connected$.pipe(
pairwise(),
map(([was, is]) => was && !is),
),
false,
);
return {
leave: () => joined$.next(false),
state$,
connected$,
reconnecting$,
disconnectReason$,
};
}
interface MembershipInputs {
joined: boolean;
error: MatrixRTCError | null;
hasTransport: boolean;
reason: MatrixDisconnectReason | null;
}
/** The next state, in priority order; `left` and `failed` are absorbing. */
function membershipState(
previous: MembershipState,
{ joined, error, hasTransport, reason }: MembershipInputs,
): MembershipState {
if (!joined) return { kind: "left" };
if (previous.kind === "failed") return previous;
if (error !== null) return { kind: "failed", error };
if (!hasTransport) return { kind: "waitingForTransport" };
if (reason === null) return { kind: "joined" };
const wasJoined =
previous.kind === "joined" || previous.kind === "reconnecting";
return wasJoined ? { kind: "reconnecting", reason } : { kind: "joining" };
}
-85
View File
@@ -1,85 +0,0 @@
/*
Copyright 2026 Element Creations Ltd.
SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
Please see LICENSE in the repository root for full details.
*/
import { describe, expect, it } from "vitest";
import { Status } from "matrix-js-sdk/lib/matrixrtc";
import { MediaConnectionState } from "../media-backend/api";
import { MatrixRTCError } from "../api";
import {
type LocalMemberState,
PublishState,
TransportState,
} from "./LocalMembership";
import { fatalError, participationStatus } from "./status";
const publishing: LocalMemberState = {
media: PublishState.Publishing,
matrix: Status.Connected,
};
const connecting: LocalMemberState = {
media: { connection: MediaConnectionState.Connecting },
matrix: Status.Connecting,
};
const error = new MatrixRTCError("gone");
describe("participationStatus", () => {
it.each<[string, LocalMemberState, boolean, boolean, boolean, string]>([
[
"no transport yet",
TransportState.Waiting,
false,
false,
false,
"waitingForTransport",
],
["joining", connecting, false, false, false, "connecting"],
["connected", publishing, true, false, false, "connected"],
[
"dropped after connecting",
connecting,
false,
true,
false,
"reconnecting",
],
["a transport error", error, false, false, false, "disconnected"],
[
"a matrix error",
{ ...publishing, matrix: error },
true,
false,
false,
"disconnected",
],
["left", publishing, true, false, true, "left"],
["left after an error", error, false, false, true, "left"],
])("%s", (_name, state, connected, reconnecting, left, expected) => {
expect(participationStatus(state, connected, reconnecting, left)).toBe(
expected,
);
});
});
describe("fatalError", () => {
it("is null while nothing is wrong", () => {
expect(fatalError(TransportState.Waiting)).toBeNull();
expect(fatalError(publishing)).toBeNull();
});
it("finds the error wherever the state holds it", () => {
expect(fatalError(error)).toBe(error);
expect(fatalError({ ...publishing, matrix: error })).toBe(error);
expect(
fatalError({ media: { connection: error }, matrix: Status.Connected }),
).toBe(error);
});
it("does not treat a publish error as fatal", () => {
expect(fatalError({ media: error, matrix: Status.Connected })).toBeNull();
});
});
-38
View File
@@ -1,38 +0,0 @@
/*
Copyright 2026 Element Creations Ltd.
SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
Please see LICENSE in the repository root for full details.
*/
import { type MatrixRTCError, type ParticipationStatus } from "../api";
import { type LocalMemberState, TransportState } from "./LocalMembership";
/** The local member's state, collapsed to what a host shows. */
export function participationStatus(
state: LocalMemberState,
connected: boolean,
reconnecting: boolean,
left: boolean,
): ParticipationStatus {
if (left) return "left";
if (fatalError(state) !== null) return "disconnected";
if (state === TransportState.Waiting) return "waitingForTransport";
if (connected) return "connected";
if (reconnecting) return "reconnecting";
return "connecting";
}
/** The error that stops the participation, if the state holds one. */
export function fatalError(state: LocalMemberState): MatrixRTCError | null {
if (state === TransportState.Waiting) return null;
if (state instanceof Error) return state;
if (state.matrix instanceof Error) return state.matrix;
if (
typeof state.media === "object" &&
"connection" in state.media &&
state.media.connection instanceof Error
)
return state.media.connection;
return null;
}
+145
View File
@@ -0,0 +1,145 @@
/*
Copyright 2026 Element Creations Ltd.
SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
Please see LICENSE in the repository root for full details.
*/
import { describe, expect, it } from "vitest";
import { MatrixRTCError, type ParticipationState } from "./api";
import { type MembershipState } from "./matrixrtc/LocalMembership";
import { MediaConnectionState } from "./media-backend/api";
import { participationState } from "./participationState";
const error = new MatrixRTCError("gone");
const waiting: ParticipationState = { kind: "waitingForTransport" };
const connecting: ParticipationState = { kind: "connecting" };
const connected: ParticipationState = { kind: "connected" };
const reconnecting: ParticipationState = {
kind: "reconnecting",
reason: "media",
};
const joined: MembershipState = { kind: "joined" };
describe("participationState", () => {
it.each<
[
string,
ParticipationState,
MembershipState,
MediaConnectionState | Error,
ParticipationState,
]
>([
[
"no transport yet",
waiting,
{ kind: "waitingForTransport" },
MediaConnectionState.Initialized,
waiting,
],
[
"joining",
waiting,
{ kind: "joining" },
MediaConnectionState.Connecting,
connecting,
],
[
"joined, media still connecting",
connecting,
joined,
MediaConnectionState.Connecting,
connecting,
],
[
"both links up",
connecting,
joined,
MediaConnectionState.Connected,
connected,
],
[
"media dropped after connecting",
connected,
joined,
MediaConnectionState.Reconnecting,
reconnecting,
],
[
"homeserver dropped after connecting",
connected,
{ kind: "reconnecting", reason: "sync" },
MediaConnectionState.Connected,
{ kind: "reconnecting", reason: "sync" },
],
[
"the matrix side wins over the media side",
reconnecting,
{ kind: "reconnecting", reason: "membership" },
MediaConnectionState.Reconnecting,
{ kind: "reconnecting", reason: "membership" },
],
[
"membership reconnecting before the media ever came up",
connecting,
{ kind: "reconnecting", reason: "sync" },
MediaConnectionState.Connecting,
connecting,
],
[
"back from a reconnect",
reconnecting,
joined,
MediaConnectionState.Connected,
connected,
],
[
"a membership error",
connected,
{ kind: "failed", error },
MediaConnectionState.Connected,
{ kind: "failed", error },
],
[
"a media connection error",
connected,
joined,
error,
{ kind: "failed", error },
],
[
"failed stays failed",
{ kind: "failed", error },
joined,
MediaConnectionState.Connected,
{ kind: "failed", error },
],
[
"left",
connected,
{ kind: "left" },
MediaConnectionState.Stopped,
{ kind: "left" },
],
[
"left after an error",
{ kind: "failed", error },
{ kind: "left" },
MediaConnectionState.Stopped,
{ kind: "left" },
],
])("%s", (_name, previous, membership, media, expected) => {
expect(participationState(previous, membership, media)).toEqual(expected);
});
it("wraps a plain media error", () => {
const state = participationState(connected, joined, new Error("boom"));
expect(state.kind).toBe("failed");
if (state.kind === "failed") {
expect(state.error).toBeInstanceOf(MatrixRTCError);
expect(state.error.cause).toBeInstanceOf(Error);
}
});
});
+42
View File
@@ -0,0 +1,42 @@
/*
Copyright 2026 Element Creations Ltd.
SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
Please see LICENSE in the repository root for full details.
*/
import { type ParticipationState } from "./api";
import { toMatrixRTCError } from "./errors";
import { type MembershipState } from "./matrixrtc/LocalMembership";
import { MediaConnectionState } from "./media-backend/api";
/**
* The membership combined with the local media connection, in priority
* order; `left` and `failed` are absorbing. "Connected once" is tracked here
* rather than in the membership because `connected` needs both links.
*/
export function participationState(
previous: ParticipationState,
membership: MembershipState,
media: MediaConnectionState | Error,
): ParticipationState {
if (membership.kind === "left") return { kind: "left" };
if (previous.kind === "failed") return previous;
if (membership.kind === "failed") return membership;
if (media instanceof Error)
return { kind: "failed", error: toMatrixRTCError(media) };
if (membership.kind === "waitingForTransport") return membership;
if (membership.kind === "joining") return { kind: "connecting" };
const reason =
membership.kind === "reconnecting"
? membership.reason
: media === MediaConnectionState.Connected
? null
: "media";
if (reason === null) return { kind: "connected" };
const wasConnected =
previous.kind === "connected" || previous.kind === "reconnecting";
return wasConnected
? { kind: "reconnecting", reason }
: { kind: "connecting" };
}