Fix planner_active_sessions gauge drifting negative

The gauge was maintained by conditional inc on "this sessionId isn't
in the in-memory `docs` Map yet" and dec on "last client for the
sessionId left." Because `docs` caches entries indefinitely — even
after every client disconnects — a reconnect found the doc still
cached, skipped the inc, then decremented again on the next
disconnect. Every reconnect-after-last-disconnect leaked one
decrement. After a day of normal reload / wifi-blip / tab-hibernate
patterns the gauge hit -182.

Replace the transitions with a `set()` derived from the live `conns`
map: count distinct session ids with at least one open WebSocket.
Self-correcting, no drift possible.

Extract `countActiveSessions()` as a pure helper + unit tests
including a regression case that simulates repeated reconnect/
disconnect cycles on one session and asserts the count never goes
negative.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Ullrich Schäfer 2026-04-22 22:37:58 +02:00
parent 6e15104496
commit 0553cf4a5e
No known key found for this signature in database
GPG key ID: A32FF691A0F752D9
2 changed files with 70 additions and 3 deletions

View file

@ -0,0 +1,45 @@
import { describe, it, expect } from "vitest";
import { countActiveSessions } from "./yjs-server.ts";
describe("countActiveSessions", () => {
it("is 0 for an empty iterable (regression: prod gauge saw -182 drift)", () => {
expect(countActiveSessions([])).toBe(0);
});
it("counts a single session with one client as 1", () => {
expect(countActiveSessions([{ sessionId: "s1" }])).toBe(1);
});
it("deduplicates multiple clients on the same session", () => {
expect(
countActiveSessions([
{ sessionId: "s1" },
{ sessionId: "s1" },
{ sessionId: "s1" },
]),
).toBe(1);
});
it("counts distinct sessions across clients", () => {
expect(
countActiveSessions([
{ sessionId: "s1" },
{ sessionId: "s2" },
{ sessionId: "s1" },
{ sessionId: "s3" },
]),
).toBe(3);
});
it("doesn't go negative under the reconnect pattern that broke the old inc/dec logic", () => {
// Simulate: client connects, client disconnects, reconnects,
// disconnects again — five times over. The old code would have
// decremented once per cycle (because `docs` stayed cached, so
// the re-connect's `isNewSession` check skipped inc). Here we
// derive from live state, so each empty snapshot is 0, not -N.
for (let i = 0; i < 5; i++) {
expect(countActiveSessions([{ sessionId: "s1" }])).toBe(1); // reconnect
expect(countActiveSessions([])).toBe(0); // disconnect
}
});
});

View file

@ -15,6 +15,29 @@ const docs = new Map<string, Y.Doc>();
const awarenessMap = new Map<string, awarenessProtocol.Awareness>();
const conns = new Map<WebSocket, { sessionId: string; clientIds: Set<number> }>();
/**
* Count the number of distinct session ids represented in an iterable
* of connection metadata. Pure, for testability the server calls it
* with the `conns` map's values.
*/
export function countActiveSessions(
connsIter: Iterable<{ sessionId: string }>,
): number {
const sessionIds = new Set<string>();
for (const { sessionId } of connsIter) sessionIds.add(sessionId);
return sessionIds.size;
}
/**
* Set the active-sessions gauge from the actual `conns` state. Derived
* each time, so it can't drift negative the previous inc/dec pattern
* leaked decrements whenever a reconnect found the doc cached in
* `docs` and skipped the matching increment.
*/
function refreshActiveSessions(): void {
plannerActiveSessions.set(countActiveSessions(conns.values()));
}
export function getOrCreateDoc(sessionId: string): Y.Doc {
let doc = docs.get(sessionId);
if (!doc) {
@ -160,13 +183,12 @@ export function setupYjsWebSocket(server: Server): WebSocketServer {
});
wss.on("connection", async (ws: WebSocket, _request: IncomingMessage, sessionId: string) => {
const isNewSession = !docs.has(sessionId);
const doc = await getOrLoadDoc(sessionId);
const awareness = getAwareness(sessionId, doc);
conns.set(ws, { sessionId, clientIds: new Set() });
plannerConnectedClients.inc();
if (isNewSession) plannerActiveSessions.inc();
refreshActiveSessions();
// Broadcast doc updates to all connections in this session
const onUpdate = (update: Uint8Array, origin: unknown) => {
@ -223,12 +245,12 @@ export function setupYjsWebSocket(server: Server): WebSocketServer {
}
conns.delete(ws);
plannerConnectedClients.dec();
refreshActiveSessions();
doc.off("update", onUpdate);
// Save when last client leaves
const hasClients = Array.from(conns.values()).some((m) => m.sessionId === sessionId);
if (!hasClients) {
plannerActiveSessions.dec();
saveSessionState(sessionId).catch(() => {});
}
});