Compare commits

...

3 Commits

Author SHA1 Message Date
Valere
53542ca957 fix missing mock for test 2025-12-05 18:28:36 +01:00
Valere
0a09b8a676 fix: race initial publish and muting 2025-12-05 18:00:33 +01:00
Valere
a3e04cecc3 add test reproducing the race 2025-12-05 17:22:48 +01:00
2 changed files with 160 additions and 19 deletions

View File

@@ -14,13 +14,19 @@ import {
type Mock,
vi,
} from "vitest";
import { ConnectionState as LivekitConenctionState } from "livekit-client";
import { type BehaviorSubject } from "rxjs";
import {
ConnectionState as LivekitConenctionState,
LocalParticipant,
type LocalTrack,
type LocalTrackPublication,
} from "livekit-client";
import { BehaviorSubject } from "rxjs";
import { logger } from "matrix-js-sdk/lib/logger";
import { ObservableScope } from "../../ObservableScope";
import { constant } from "../../Behavior";
import {
flushPromises,
mockLivekitRoom,
mockLocalParticipant,
mockMediaDevices,
@@ -33,8 +39,15 @@ import {
import { type MuteStates } from "../../MuteStates";
import { FailToStartLivekitConnection } from "../../../utils/errors";
let scope: ObservableScope;
beforeEach(() => {
scope = new ObservableScope();
});
afterEach(() => scope.end());
describe("Publisher", () => {
let scope: ObservableScope;
let connection: Connection;
let muteStates: MuteStates;
beforeEach(() => {
@@ -50,7 +63,6 @@ describe("Publisher", () => {
setHandler: vi.fn(),
},
} as unknown as MuteStates;
scope = new ObservableScope();
connection = {
state$: constant({
state: "ConnectedToLkRoom",
@@ -62,8 +74,6 @@ describe("Publisher", () => {
} as unknown as Connection;
});
afterEach(() => scope.end());
it("throws if livekit room could not publish", async () => {
const publisher = new Publisher(
scope,
@@ -88,14 +98,19 @@ describe("Publisher", () => {
(
connection.livekitRoom.localParticipant.createTracks as Mock
).mockResolvedValue([{}, {}]);
).mockResolvedValue([
{
kind: "audio",
mute: vi.fn(),
},
]);
await expect(publisher.createAndSetupTracks()).resolves.not.toThrow();
expect(
connection.livekitRoom.localParticipant.createTracks,
).toHaveBeenCalledOnce();
// failiour due to localParticipant.publishTrack
// failure due to localParticipant.publishTrack
(
connection.livekitRoom.localParticipant.publishTrack as Mock
).mockRejectedValue(Error("testError"));
@@ -104,7 +119,7 @@ describe("Publisher", () => {
new FailToStartLivekitConnection("testError"),
);
// does not try other conenction after the first one failed
// does not try other connection after the first one failed
expect(
connection.livekitRoom.localParticipant.publishTrack,
).toHaveBeenCalledTimes(1);
@@ -135,6 +150,111 @@ describe("Publisher", () => {
expect(
connection.livekitRoom.localParticipant.publishTrack,
).toHaveBeenCalledTimes(3);
).toHaveBeenCalledTimes(2);
});
});
describe("Bug fix", () => {
// There is a race condition when creating and publishing tracks while the mute state changes.
// This race condition could cause tracks to be published even though they are muted at the
// beginning of a call coming from lobby.
// This is caused by our stack using manually the low level API to create and publish tracks,
// but also using the higher level setMicrophoneEnabled and setCameraEnabled functions that also create
// and publish tracks, and managing pending publications.
// Race is as follow, on creation of the Publisher we create the tracks then publish them.
// If in the middle of that process the mute state changes:
// - the `setMicrophoneEnabled` will be no-op because it is not aware of our created track and can't see any pending publication
// - If start publication is requested it will publish the track even though there was a mute request.
it("wrongly publish tracks while muted", async () => {
const audioEnabled$ = new BehaviorSubject(true);
const muteStates = {
audio: {
enabled$: audioEnabled$,
unsetHandler: vi.fn(),
setHandler: vi.fn(),
},
video: {
enabled$: constant(false),
unsetHandler: vi.fn(),
setHandler: vi.fn(),
},
} as unknown as MuteStates;
const mockSendDataPacket = vi.fn();
const mockEngine = {
client: {
sendUpdateLocalMetadata: vi.fn(),
},
on: vi.fn().mockReturnThis(),
sendDataPacket: mockSendDataPacket,
};
// cont mockRoomOptions = {} as InternalRoomOptions;
const localParticipant = new LocalParticipant(
"local-sid",
"local-identity",
// @ts-expect-error - for that test we want a real LocalParticipant to have the pending publications logic
mockEngine,
{},
new Map(),
{},
);
const connection = {
state$: constant({
state: "ConnectedToLkRoom",
livekitConnectionState$: constant(LivekitConenctionState.Connected),
}),
livekitRoom: mockLivekitRoom({
localParticipant,
}),
} as unknown as Connection;
const mediaDevices = mockMediaDevices({});
const mockTrack = vi.mocked<LocalTrack>({
kind: "audio",
mute: vi.fn(),
} as Partial<LocalTrack> as LocalTrack);
const createTrackLock = Promise.withResolvers<void>();
const createTrackSpy = vi.spyOn(localParticipant, "createTracks");
createTrackSpy.mockImplementation(async () => {
await createTrackLock.promise;
return [mockTrack];
});
const publishTrackSpy = vi.spyOn(localParticipant, "publishTrack");
publishTrackSpy.mockResolvedValue({} as unknown as LocalTrackPublication);
const publisher = new Publisher(
scope,
connection,
mediaDevices,
muteStates,
constant({ supported: false, processor: undefined }),
logger,
);
// Initially the audio is unmuted, so creating tracks should publish the audio track
const createTracks = publisher.createAndSetupTracks();
publisher.tracks$.subscribe(() => {
void publisher.startPublishing();
});
// now mute the audio before allowing track creation to complete
audioEnabled$.next(false);
// const publishing = publisher.startPublishing();
createTrackLock.resolve();
await createTracks;
// await publishing;
await flushPromises();
// It should not publish or instead call track.mute()
try {
expect(publishTrackSpy).not.toHaveBeenCalled();
} catch {
expect(mockTrack.mute).toHaveBeenCalled();
}
});
});

View File

@@ -144,6 +144,7 @@ export class Publisher {
this.logger.error("Failed to create tracks", error);
});
}
// TODO why throw here? should we just do nothing?
throw Error("audio and video is false");
}
@@ -184,16 +185,36 @@ export class Publisher {
for (const track of this.tracks$.value) {
this.logger.info("publish ", this.tracks$.value.length, "tracks");
// TODO: handle errors? Needs the signaling connection to be up, but it has some retries internally
// with a timeout.
await lkRoom.localParticipant.publishTrack(track).catch((error) => {
this.logger.error("Failed to publish track", error);
throw new FailToStartLivekitConnection(
error instanceof Error ? error.message : error,
);
});
this.logger.info("published track ", track.kind, track.id);
// XXXX: Patch: Check if the track has been muted manually before publishing
// This is only a patch, the proper way would be to use livekit high-level enabled/disabled APIs
// or only use the low level create/publish APIs and have our own pending publication protection.
// Maybe we could change the livekit api to pre-load tracks without publishing them yet?
// Are we sure this is needed at all? What are the gains?
let isEnabled: boolean;
if (track.kind === Track.Kind.Audio) {
isEnabled = this.muteStates.audio.enabled$.value;
} else if (track.kind === Track.Kind.Video) {
isEnabled = this.muteStates.video.enabled$.value;
} else {
throw new Error("Unsupported track kind " + track.kind);
}
if (!isEnabled) {
// TODO should we also drop it?
// I believe the high-level LiveKit APIs will recreate a track?
await track.mute();
} else {
// TODO: handle errors? Needs the signaling connection to be up, but it has some retries internally
// with a timeout.
await lkRoom.localParticipant.publishTrack(track).catch((error) => {
this.logger.error("Failed to publish track", error);
throw new FailToStartLivekitConnection(
error instanceof Error ? error.message : error,
);
});
this.logger.info("published track ", track.kind, track.id);
}
// TODO: check if the connection is still active? and break the loop if not?
}
this._publishing$.next(true);