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.

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(

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
+ * with their tracks and subscriptions: those of a server that crashed or could not flush on shutdown, and those the
- * sessions stay open, also while Redis restarts.
+ * 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. */
+    /** A session without statistics for this long is a candidate (3 × the statistics interval). */
-    const REDIS_RESTART_WATCH_SECONDS = 240;
+    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(() => {
+    beforeEach(() => {
-        for (const pid of pids) {
+        caseStartedAt = new Date();
-            killProcess(pid, "SIGKILL");
+        passed = false;
+    });
+
+    afterEach(async () => {
+        dropClients();
+        LocalDeployment.saveServerLogs(caseStartedAt);
+        if (!passed) {
+            LocalDeployment.stop();
+            await LocalDeployment.start();
         }
-        pids = [];
+    }, 240000);
-        LocalDeployment.stop();
-    }, 120000);
 
     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<string[]> {
-        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<Date> {
+        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<string> {
+        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({
+        const room = await mongo.findEntity(Collection.ROOMS, { name: roomName });
-            status: "closed",
+        expect(room).toMatchObject({ status: "closed", closedBy: "reconciler", endReason: "RECONCILED" });
-            closedBy: "reconciler",
+        // The room never ends before its sessions: it ends with the last one.
-            endReason: "RECONCILED"
+        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(
+        expect(await mongo.count(Collection.PARTICIPANTS, { _id: { $in: sids }, closedBy: "event" })).toBe(sids.length);
-            await mongo.count(Collection.PARTICIPANTS, { _id: { $in: sids }, closedBy: "event" })
+        passed = true;
-        ).toBe(sids.length);
     }, 900000);
 });

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;
+        }
+    }
 }

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<Date> {
         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 (LocalDeployment.serverLogsSince(since).includes("analytics database initialized")) {
-            if (logs.includes("analytics database initialized")) {
+                return new Date();
-                return;
             }
-            await sleep(1);
+            await sleep(0.5);
         }
 
         throw new Error("The OpenVidu server did not initialize its analytics database");

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<void> {
+        await this.db.collection<AnalyticsDocument>(collection).insertOne(doc);
+    }
+
     public async findEvents(filter: Query): Promise<Document[]> {
         return this.db.collection<AnalyticsDocument>(Collection.EVENTS).find(filter).sort({ ts: 1 }).toArray();
     }