From e40f171278ec1f06770a639418b478e4e2f92f6c Mon Sep 17 00:00:00 2001 From: juancarmore Date: Mon, 2 Nov 2026 20:15:49 +0100 Subject: [PATCH] openvidu-test-integration: reconciler cases for the startup run, the live listing, a planned restart and a Redis without its data The cases share one deployment, the Redis watch lasts 150 s and the MongoDB outage case waits for the session instead of fixed sleeps. --- .../tests/analytics-mongo-outage.test.ts | 7 +- .../tests/analytics-reconciler.test.ts | 259 ++++++++++++++++-- .../tests/utils/livekit.ts | 13 + .../tests/utils/local-deployment.ts | 61 ++++- .../tests/utils/mongodb.ts | 5 + 5 files changed, 301 insertions(+), 44 deletions(-) diff --git a/openvidu-test-integration/tests/analytics-mongo-outage.test.ts b/openvidu-test-integration/tests/analytics-mongo-outage.test.ts index 1f2ea3d0d..339fc9d7c 100644 --- a/openvidu-test-integration/tests/analytics-mongo-outage.test.ts +++ b/openvidu-test-integration/tests/analytics-mongo-outage.test.ts @@ -34,10 +34,13 @@ describe("OpenVidu analytics during a MongoDB outage", () => { LocalDeployment.stopContainer("mongo"); pids = [Livekit.joinParticipantToRoom(identity, roomName)]; - await sleep(20); + await waitFor(async () => Livekit.isParticipantInRoom(identity, roomName), 60, "the session to join"); + await sleep(5); killProcess(pids[0]); pids = []; - await sleep(20); + await waitFor(async () => !Livekit.isParticipantInRoom(identity, roomName), 60, "the session to leave"); + // A few batch ticks of the writer fail while MongoDB is still down, so the leave event is retried. + await sleep(5); LocalDeployment.startContainer("mongo"); await waitFor( diff --git a/openvidu-test-integration/tests/analytics-reconciler.test.ts b/openvidu-test-integration/tests/analytics-reconciler.test.ts index 1e24c8560..2d995cfa4 100644 --- a/openvidu-test-integration/tests/analytics-reconciler.test.ts +++ b/openvidu-test-integration/tests/analytics-reconciler.test.ts @@ -1,13 +1,14 @@ import { killProcess, sleep, waitFor } from "./utils/helper"; import { Livekit } from "./utils/livekit"; import { LocalDeployment } from "./utils/local-deployment"; -import { MongoService } from "./utils/mongodb"; +import { AnalyticsDocument, MongoService } from "./utils/mongodb"; import { Collection } from "./utils/types"; /** * The reconciler of the OpenVidu server closes the rooms and participant sessions whose end event never arrives, - * with their tracks and subscriptions: those of a server that crashed or could not flush on shutdown. Healthy - * sessions stay open, also while Redis restarts. + * with their tracks and subscriptions: those of a server that crashed or could not flush on shutdown, and those the + * store holds open while LiveKit no longer has them. Healthy sessions stay open, also while Redis restarts. The cases + * share one deployment; a failed case hands a fresh one to the next. */ describe("OpenVidu analytics reconciler", () => { const mongo = MongoService.getInstance(); @@ -17,29 +18,48 @@ describe("OpenVidu analytics reconciler", () => { const ROOM_CLOSURE_TIMEOUT_SECONDS = 420; /** Statistics are flushed every 30 s: the estimated end time is at most this early. */ const STATS_INTERVAL_MS = 30_000; - /** How long healthy sessions are watched after Redis restarts. */ - const REDIS_RESTART_WATCH_SECONDS = 240; + /** A session without statistics for this long is a candidate (3 × the statistics interval). */ + const STALE_AFTER_MS = 3 * STATS_INTERVAL_MS; + /** The reconciler period the CI workflow configures. */ + const RECONCILER_INTERVAL_MS = 10_000; + /** + * How long healthy sessions are watched after Redis restarts: longer than a session takes to become a candidate, + * plus two reconciler runs and one more statistics flush. + */ + const REDIS_RESTART_WATCH_SECONDS = 150; + /** A candidate of an alive node is closed after two consecutive runs that do not find it in LiveKit. */ + const ABSENT_CLOSURE_TIMEOUT_SECONDS = 60; let pids: number[] = []; + let caseStartedAt = new Date(); + let passed = false; - beforeEach(async () => { + beforeAll(async () => { await LocalDeployment.start(); }, 120000); - afterEach(() => { - for (const pid of pids) { - killProcess(pid, "SIGKILL"); + beforeEach(() => { + caseStartedAt = new Date(); + passed = false; + }); + + afterEach(async () => { + dropClients(); + LocalDeployment.saveServerLogs(caseStartedAt); + if (!passed) { + LocalDeployment.stop(); + await LocalDeployment.start(); } - pids = []; - LocalDeployment.stop(); - }, 120000); + }, 240000); afterAll(async () => { + // Each case saved its own part of the log. + LocalDeployment.stop(false); await mongo.disconnect(); - }); + }, 120000); /** Joins publishers to a room and waits until every session is active with statistics; returns their sids. */ async function startRoomWithPublishers(roomName: string, identities: string[]): Promise { - pids = identities.map((identity) => Livekit.joinParticipantToRoom(identity, roomName)); + pids.push(...identities.map((identity) => Livekit.joinParticipantToRoom(identity, roomName))); await waitFor( async () => (await mongo.count(Collection.PARTICIPANTS, { @@ -51,7 +71,11 @@ describe("OpenVidu analytics reconciler", () => { 120, `the sessions of '${roomName}' to be active with statistics` ); - const participants = await mongo.findEntities(Collection.PARTICIPANTS, { roomName, status: "active" }); + const participants = await mongo.findEntities(Collection.PARTICIPANTS, { + roomName, + identity: { $in: identities }, + status: "active" + }); return participants.map((participant) => participant._id); } @@ -63,11 +87,18 @@ describe("OpenVidu analytics reconciler", () => { pids = []; } + /** Waits for the server to be back after a restart that began at `since`; returns when it was seen ready. */ + async function serverReady(since: Date): Promise { + LocalDeployment.startContainer("openvidu"); + return LocalDeployment.waitForAnalytics(since, 120); + } + /** * Every session closed by the reconciler: left with the reason RECONCILED, an end time at most one statistics - * interval before the failure, one synthetic event each, and no track or subscription of theirs left open. + * interval before the failure, one synthetic event each naming the closure reason, and no track or subscription + * of theirs left open. */ - async function expectClosedByReconciler(sids: string[], failedAt: Date) { + async function expectClosedByReconciler(sids: string[], failedAt: Date, reason = "node-dead") { for (const sid of sids) { const participant = await mongo.findEntity(Collection.PARTICIPANTS, { _id: sid }); expect(participant).toMatchObject({ status: "left", closedBy: "reconciler", disconnectReason: "RECONCILED" }); @@ -77,7 +108,8 @@ describe("OpenVidu analytics reconciler", () => { expect(await mongo.findSyntheticEvent(sid)).toMatchObject({ type: "PARTICIPANT_LEFT", participantSid: sid, - synthetic: true + synthetic: true, + payload: { reconciler: { reason } } }); } expect(await mongo.countOpen(Collection.TRACKS, { participantSid: { $in: sids } })).toBe(0); @@ -88,6 +120,28 @@ describe("OpenVidu analytics reconciler", () => { ).toBe(0); } + /** + * Inserts an active session in the room of `template`, on the same node, whose statistics stopped 100 s ago and + * that LiveKit never had: what a lost leave event leaves behind on a server that is still alive. + */ + async function seedStaleParticipant(template: AnalyticsDocument, identity: string): Promise { + const sid = `PA_${identity.replace(/[^A-Za-z0-9]/g, "")}${Date.now()}`; + const joinedAt = new Date(Date.now() - 120_000); + const { leftAt, durationMs, disconnectReason, closedBy, internal, ...fields } = template; + await mongo.insertEntity(Collection.PARTICIPANTS, { + ...fields, + _id: sid, + identity, + identityLower: identity.toLowerCase(), + status: "active", + joinedAt, + activeAt: joinedAt, + lastEventAt: joinedAt, + lastStatsAt: new Date(Date.now() - 100_000) + }); + return sid; + } + it("closes the sessions and the room of a crashed server", async () => { const roomName = "ReconcilerCrash"; const sids = await startRoomWithPublishers(roomName, ["crash-alice", "crash-bob"]); @@ -96,7 +150,7 @@ describe("OpenVidu analytics reconciler", () => { LocalDeployment.killContainer("openvidu"); dropClients(); await sleep(5); - LocalDeployment.startContainer("openvidu"); + await serverReady(crashedAt); await waitFor( async () => (await mongo.countOpen(Collection.PARTICIPANTS, { _id: { $in: sids } })) === 0, @@ -109,11 +163,14 @@ describe("OpenVidu analytics reconciler", () => { ROOM_CLOSURE_TIMEOUT_SECONDS, "the room of the crashed server to be closed" ); - expect(await mongo.findEntity(Collection.ROOMS, { name: roomName })).toMatchObject({ - status: "closed", - closedBy: "reconciler", - endReason: "RECONCILED" - }); + const room = await mongo.findEntity(Collection.ROOMS, { name: roomName }); + expect(room).toMatchObject({ status: "closed", closedBy: "reconciler", endReason: "RECONCILED" }); + // The room never ends before its sessions: it ends with the last one. + const leftAts = (await mongo.findEntities(Collection.PARTICIPANTS, { _id: { $in: sids } })).map((participant) => + (participant.leftAt as Date).getTime() + ); + expect((room!.endedAt as Date).getTime()).toBe(Math.max(...leftAts)); + passed = true; }, 900000); it("closes the sessions that a server stopped before its flush could not write", async () => { @@ -123,7 +180,7 @@ describe("OpenVidu analytics reconciler", () => { const stoppedAt = new Date(); LocalDeployment.stopContainerWithTimeout("openvidu", 1); dropClients(); - LocalDeployment.startContainer("openvidu"); + await serverReady(stoppedAt); await waitFor( async () => (await mongo.countOpen(Collection.PARTICIPANTS, { _id: { $in: sids } })) === 0, @@ -136,17 +193,147 @@ describe("OpenVidu analytics reconciler", () => { ).map((participant) => participant._id); await expectClosedByReconciler(reconciled, stoppedAt); expect(await mongo.countOpen(Collection.TRACKS, { participantSid: { $in: sids } })).toBe(0); + passed = true; }, 900000); - it("keeps healthy sessions open while Redis restarts, and records their real leave events", async () => { + it("closes the sessions of a server that was down for longer than the stale threshold as soon as it starts", async () => { + const sids = await startRoomWithPublishers("ReconcilerStartup", ["startup-alice", "startup-bob"]); + + // A stop without any grace is a crash that keeps the server down: its sessions are stale when it starts. + const crashedAt = new Date(); + LocalDeployment.stopContainerWithTimeout("openvidu", 0); + dropClients(); + await sleep((STALE_AFTER_MS + STATS_INTERVAL_MS) / 1000); + const readyAt = await serverReady(crashedAt); + + // The startup run closes them; the first periodic run would come one reconciler period later. + let closedAt: Date | undefined; + const deadline = readyAt.getTime() + CLOSURE_TIMEOUT_SECONDS * 1000; + while (!closedAt && Date.now() < deadline) { + if ((await mongo.countOpen(Collection.PARTICIPANTS, { _id: { $in: sids } })) === 0) { + closedAt = new Date(); + } else { + await sleep(0.5); + } + } + expect(closedAt).toBeDefined(); + expect(closedAt!.getTime() - readyAt.getTime()).toBeLessThan(RECONCILER_INTERVAL_MS - 2_000); + await expectClosedByReconciler(sids, crashedAt); + passed = true; + }, 900000); + + it("closes the sessions and the rooms that an alive server's LiveKit no longer has", async () => { + const roomName = "ReconcilerAbsent"; + const [realSid] = await startRoomWithPublishers(roomName, ["absent-alice"]); + const participant = (await mongo.findEntity(Collection.PARTICIPANTS, { _id: realSid }))!; + const room = (await mongo.findEntity(Collection.ROOMS, { name: roomName, status: "active" }))!; + + // A session of the live room that LiveKit never listed, and an empty room past its empty timeout that LiveKit + // never had, both of the alive server. + const ghostSid = await seedStaleParticipant(participant, "absent-ghost"); + const emptyRoomSid = `RM_ReconcilerAbsentEmpty${Date.now()}`; + const lastEventAt = new Date(Date.now() - 8 * 60_000); + const { endedAt, durationMs, endReason, closedBy, internal, ...roomFields } = room; + await mongo.insertEntity(Collection.ROOMS, { + ...roomFields, + _id: emptyRoomSid, + name: "ReconcilerAbsentEmpty", + nameLower: "reconcilerabsentempty", + createdAt: new Date(lastEventAt.getTime() - 60_000), + lastEventAt, + lastStatsAt: lastEventAt, + participants: { sessions: 0, unique: 0, active: 0, peakActive: 0, aborted: 0, byKind: {} } + }); + + await waitFor( + async () => + (await mongo.countOpen(Collection.PARTICIPANTS, { _id: ghostSid })) === 0 && + (await mongo.countOpen(Collection.ROOMS, { _id: emptyRoomSid })) === 0, + ABSENT_CLOSURE_TIMEOUT_SECONDS, + "the entities LiveKit does not have to be closed" + ); + expect(await mongo.findEntity(Collection.PARTICIPANTS, { _id: ghostSid })).toMatchObject({ + status: "left", + closedBy: "reconciler", + disconnectReason: "RECONCILED" + }); + expect(await mongo.findSyntheticEvent(ghostSid)).toMatchObject({ + type: "PARTICIPANT_LEFT", + payload: { reconciler: { reason: "absent-live" } } + }); + expect(await mongo.findEntity(Collection.ROOMS, { _id: emptyRoomSid })).toMatchObject({ + status: "closed", + closedBy: "reconciler", + endReason: "RECONCILED", + endedAt: lastEventAt + }); + expect(await mongo.findSyntheticEvent(emptyRoomSid)).toMatchObject({ + type: "ROOM_ENDED", + payload: { reconciler: { reason: "absent-live" } } + }); + // The sessions LiveKit lists stay open. + expect(await mongo.countOpen(Collection.PARTICIPANTS, { _id: realSid })).toBe(1); + expect(await mongo.countOpen(Collection.ROOMS, { _id: room._id })).toBe(1); + passed = true; + }, 900000); + + it("closes the rooms and sessions of a planned restart with their real end events", async () => { + const roomName = "ReconcilerPlannedRestart"; + const sids = await startRoomWithPublishers(roomName, ["planned-alice", "planned-bob"]); + const startedAt = LocalDeployment.containerStartedAt("openvidu"); + + // The first SIGTERM makes the server drain; the second one closes its rooms and it exits after its flush. + const stoppingAt = new Date(); + LocalDeployment.signalContainer("openvidu", "SIGTERM"); + await sleep(3); + LocalDeployment.signalContainer("openvidu", "SIGTERM"); + dropClients(); + await waitFor( + async () => + !LocalDeployment.isContainerRunning("openvidu") || + LocalDeployment.containerStartedAt("openvidu") !== startedAt, + 60, + "the server to exit" + ); + await serverReady(stoppingAt); + + const room = await mongo.findEntity(Collection.ROOMS, { name: roomName }); + expect(room).toMatchObject({ status: "closed", closedBy: "event", endReason: "SERVER_SHUTDOWN" }); + for (const sid of sids) { + const participant = await mongo.findEntity(Collection.PARTICIPANTS, { _id: sid }); + expect(participant).toMatchObject({ status: "left", closedBy: "event" }); + // The final statistics were flushed on the way out. + expect((participant!.lastStatsAt as Date).getTime()).toBeGreaterThanOrEqual(stoppingAt.getTime()); + expect(await mongo.findSyntheticEvent(sid)).toBeNull(); + } + const types = (await mongo.findEvents({ roomName })).map((event) => event.type as string); + expect(types).toContain("ROOM_ENDED"); + expect(types.filter((type) => type === "PARTICIPANT_LEFT")).toHaveLength(sids.length); + passed = true; + }, 900000); + + it("keeps healthy sessions open while Redis restarts and loses its state, trusts LiveKit again once it lists rooms, and records the real leave events", async () => { const roomName = "ReconcilerRedis"; const sids = await startRoomWithPublishers(roomName, ["redis-alice", "redis-bob"]); + const template = (await mongo.findEntity(Collection.PARTICIPANTS, { _id: sids[0] }))!; const runBefore = await mongo.schemaMeta("reconciler"); + // The deployment's Redis persists its data, so it comes back with LiveKit's listing; emptying it is a Redis + // restarted without its data. const restartedAt = new Date(); LocalDeployment.restartContainer("redis"); + LocalDeployment.flushRedis(); + // A stale session of the live room: while LiveKit lists no room and the room still receives statistics, the + // listing is not trusted and the session is kept. + const ghostSid = await seedStaleParticipant(template, "redis-ghost"); + await waitFor( + async () => LocalDeployment.serverLogsSince(restartedAt).includes("live-state-untrusted"), + 60, + "a reconciler run that does not trust the empty live listing" + ); + expect(await mongo.countOpen(Collection.PARTICIPANTS, { _id: ghostSid })).toBe(1); - const deadline = Date.now() + REDIS_RESTART_WATCH_SECONDS * 1000; + const deadline = restartedAt.getTime() + REDIS_RESTART_WATCH_SECONDS * 1000; while (Date.now() < deadline) { expect(await mongo.countOpen(Collection.PARTICIPANTS, { _id: { $in: sids } })).toBe(sids.length); await sleep(10); @@ -157,6 +344,19 @@ describe("OpenVidu analytics reconciler", () => { Math.max(restartedAt.getTime(), (runBefore?.lastRunAt as Date | undefined)?.getTime() ?? 0) ); + // A new room puts LiveKit's listing back in Redis: the listing is trusted again and the stale session, which + // it does not list, is closed. + pids.push(Livekit.joinParticipantToRoom("redis-carol", "ReconcilerRedisAfter")); + await waitFor( + async () => (await mongo.countOpen(Collection.PARTICIPANTS, { _id: ghostSid })) === 0, + ABSENT_CLOSURE_TIMEOUT_SECONDS, + "the stale session to be closed once the listing is trusted" + ); + expect(await mongo.findSyntheticEvent(ghostSid)).toMatchObject({ + payload: { reconciler: { reason: "absent-live" } } + }); + expect(await mongo.countOpen(Collection.PARTICIPANTS, { _id: { $in: sids } })).toBe(sids.length); + for (const pid of pids) { killProcess(pid); } @@ -166,8 +366,7 @@ describe("OpenVidu analytics reconciler", () => { 120, "the sessions to be closed by their leave events" ); - expect( - await mongo.count(Collection.PARTICIPANTS, { _id: { $in: sids }, closedBy: "event" }) - ).toBe(sids.length); + expect(await mongo.count(Collection.PARTICIPANTS, { _id: { $in: sids }, closedBy: "event" })).toBe(sids.length); + passed = true; }, 900000); }); diff --git a/openvidu-test-integration/tests/utils/livekit.ts b/openvidu-test-integration/tests/utils/livekit.ts index ec37db4a6..7fc36f730 100644 --- a/openvidu-test-integration/tests/utils/livekit.ts +++ b/openvidu-test-integration/tests/utils/livekit.ts @@ -1,3 +1,4 @@ +import { execSync } from "child_process"; import { execCommandInBackground } from "./helper"; const LIVEKIT_URL = "ws://localhost:7880"; @@ -30,4 +31,16 @@ export class Livekit { } return pid; } + + /** Whether LiveKit lists a participant with this identity in the room; false when the room does not exist. */ + static isParticipantInRoom(participantIdentity: string, roomName: string): boolean { + const args = ["--url", LIVEKIT_URL, "--api-key", LIVEKIT_API_KEY, "--api-secret", LIVEKIT_API_SECRET]; + try { + return execSync(`${LK_CLI} room participants list ${args.join(" ")} ${roomName}`, { stdio: "pipe" }) + .toString() + .includes(participantIdentity); + } catch { + return false; + } + } } diff --git a/openvidu-test-integration/tests/utils/local-deployment.ts b/openvidu-test-integration/tests/utils/local-deployment.ts index c40994adb..39ce0cdf6 100644 --- a/openvidu-test-integration/tests/utils/local-deployment.ts +++ b/openvidu-test-integration/tests/utils/local-deployment.ts @@ -20,17 +20,29 @@ export class LocalDeployment { console.log("Local deployment started"); } - static stop() { + /** Removes the deployment, saving the server log first unless the caller already saved what it needs. */ + static stop(saveLogs = true) { console.log("Stopping local deployment..."); - LocalDeployment.saveServerLogs(); + if (saveLogs) { + LocalDeployment.saveServerLogs(); + } execCommand(`docker compose -f ${LOCAL_DEPLOYMENT_PATH} down -v`); } - /** Saves the log of the OpenVidu server in logs/, named after the current test, before the deployment is removed. */ - private static saveServerLogs() { + /** + * Saves the log of the OpenVidu server in logs/, named after the current test, before the deployment is removed; + * with `since`, only what the server logged from then on. + */ + static saveServerLogs(since?: Date) { const name = (expect.getState().currentTestName ?? "deployment").replace(/[^\w-]+/g, "-"); + const sinceArg = since ? `--since ${since.toISOString()}` : ""; mkdirSync(LOGS_PATH, { recursive: true }); - execCommand(`docker logs openvidu > ${join(LOGS_PATH, `${name}.log`)} 2>&1 || true`); + execCommand(`docker logs ${sinceArg} openvidu > ${join(LOGS_PATH, `${name}.log`)} 2>&1 || true`); + } + + /** What the OpenVidu server logged since the given time. */ + static serverLogsSince(since: Date): string { + return execCommand(`docker logs --since ${since.toISOString()} openvidu 2>&1 || true`); } static startContainer(containerName: string) { @@ -55,22 +67,47 @@ export class LocalDeployment { execCommand(`docker kill ${containerName}`); } + /** Sends a signal to the main process of a container, as an orchestrator does on a planned stop. */ + static signalContainer(containerName: string, signal: NodeJS.Signals) { + console.log(`Sending ${signal} to container ${containerName}...`); + execCommand(`docker kill --signal=${signal} ${containerName}`); + } + + /** When the current run of a container started, to tell a restarted container from the previous run. */ + static containerStartedAt(containerName: string): string { + return execCommand(`docker inspect -f {{.State.StartedAt}} ${containerName}`); + } + + static isContainerRunning(containerName: string): boolean { + return execCommand(`docker inspect -f {{.State.Running}} ${containerName}`) === "true"; + } + + /** + * Empties Redis, as a Redis restarted without its persisted data: LiveKit's room listing is gone while the rooms + * keep running on the server. + */ + static flushRedis() { + console.log("Flushing Redis..."); + execCommand("docker exec redis redis-cli --no-auth-warning -a redispassword FLUSHALL"); + } + static restartContainer(containerName: string) { console.log(`Restarting container ${containerName}...`); execCommand(`docker restart ${containerName}`); } - /** Waits until the OpenVidu server logs that its analytics database is initialized after the given time. */ - static async waitForAnalytics(since: Date, timeoutSeconds = 60) { + /** + * Waits until the OpenVidu server logs that its analytics database is initialized after the given time, and + * returns when that was seen, polling every half second. + */ + static async waitForAnalytics(since: Date, timeoutSeconds = 60): Promise { const deadline = Date.now() + timeoutSeconds * 1000; - const sinceArg = since.toISOString(); while (Date.now() < deadline) { - const logs = execCommand(`docker logs --since ${sinceArg} openvidu 2>&1 || true`); - if (logs.includes("analytics database initialized")) { - return; + if (LocalDeployment.serverLogsSince(since).includes("analytics database initialized")) { + return new Date(); } - await sleep(1); + await sleep(0.5); } throw new Error("The OpenVidu server did not initialize its analytics database"); diff --git a/openvidu-test-integration/tests/utils/mongodb.ts b/openvidu-test-integration/tests/utils/mongodb.ts index 688b79a4d..2b7dd9f72 100644 --- a/openvidu-test-integration/tests/utils/mongodb.ts +++ b/openvidu-test-integration/tests/utils/mongodb.ts @@ -67,6 +67,11 @@ export class MongoService { return this.count(collection, { ...filter, status: { $in: OPEN_STATUSES[collection] ?? [] } }); } + /** Inserts a document as the writer would have left it, for the cases that need a state no failure can produce. */ + public async insertEntity(collection: Collection, doc: AnalyticsDocument): Promise { + await this.db.collection(collection).insertOne(doc); + } + public async findEvents(filter: Query): Promise { return this.db.collection(Collection.EVENTS).find(filter).sort({ ts: 1 }).toArray(); }