Skip to content

Commit 0dec234

Browse files
steipetesteipete
committed
perf(logging): split diagnostic session state module
1 parent bbe3b2b commit 0dec234

3 files changed

Lines changed: 119 additions & 101 deletions

File tree

‎src/logging/diagnostic-session-state.ts‎

Lines changed: 91 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,91 @@
1+
export type SessionStateValue = "idle" | "processing" | "waiting";
2+
3+
export type SessionState = {
4+
sessionId?: string;
5+
sessionKey?: string;
6+
lastActivity: number;
7+
state: SessionStateValue;
8+
queueDepth: number;
9+
};
10+
11+
export type SessionRef = {
12+
sessionId?: string;
13+
sessionKey?: string;
14+
};
15+
16+
export const diagnosticSessionStates = new Map<string, SessionState>();
17+
18+
const SESSION_STATE_TTL_MS = 30 * 60 * 1000;
19+
const SESSION_STATE_PRUNE_INTERVAL_MS = 60 * 1000;
20+
const SESSION_STATE_MAX_ENTRIES = 2000;
21+
22+
let lastSessionPruneAt = 0;
23+
24+
export function pruneDiagnosticSessionStates(now = Date.now(), force = false): void {
25+
const shouldPruneForSize = diagnosticSessionStates.size > SESSION_STATE_MAX_ENTRIES;
26+
if (!force && !shouldPruneForSize && now - lastSessionPruneAt < SESSION_STATE_PRUNE_INTERVAL_MS) {
27+
return;
28+
}
29+
lastSessionPruneAt = now;
30+
31+
for (const [key, state] of diagnosticSessionStates.entries()) {
32+
const ageMs = now - state.lastActivity;
33+
const isIdle = state.state === "idle";
34+
if (isIdle && state.queueDepth <= 0 && ageMs > SESSION_STATE_TTL_MS) {
35+
diagnosticSessionStates.delete(key);
36+
}
37+
}
38+
39+
if (diagnosticSessionStates.size <= SESSION_STATE_MAX_ENTRIES) {
40+
return;
41+
}
42+
const excess = diagnosticSessionStates.size - SESSION_STATE_MAX_ENTRIES;
43+
const ordered = Array.from(diagnosticSessionStates.entries()).toSorted(
44+
(a, b) => a[1].lastActivity - b[1].lastActivity,
45+
);
46+
for (let i = 0; i < excess; i += 1) {
47+
const key = ordered[i]?.[0];
48+
if (!key) {
49+
break;
50+
}
51+
diagnosticSessionStates.delete(key);
52+
}
53+
}
54+
55+
function resolveSessionKey({ sessionKey, sessionId }: SessionRef) {
56+
return sessionKey ?? sessionId ?? "unknown";
57+
}
58+
59+
export function getDiagnosticSessionState(ref: SessionRef): SessionState {
60+
pruneDiagnosticSessionStates();
61+
const key = resolveSessionKey(ref);
62+
const existing = diagnosticSessionStates.get(key);
63+
if (existing) {
64+
if (ref.sessionId) {
65+
existing.sessionId = ref.sessionId;
66+
}
67+
if (ref.sessionKey) {
68+
existing.sessionKey = ref.sessionKey;
69+
}
70+
return existing;
71+
}
72+
const created: SessionState = {
73+
sessionId: ref.sessionId,
74+
sessionKey: ref.sessionKey,
75+
lastActivity: Date.now(),
76+
state: "idle",
77+
queueDepth: 0,
78+
};
79+
diagnosticSessionStates.set(key, created);
80+
pruneDiagnosticSessionStates(Date.now(), true);
81+
return created;
82+
}
83+
84+
export function getDiagnosticSessionStateCountForTest(): number {
85+
return diagnosticSessionStates.size;
86+
}
87+
88+
export function resetDiagnosticSessionStateForTest(): void {
89+
diagnosticSessionStates.clear();
90+
lastSessionPruneAt = 0;
91+
}

‎src/logging/diagnostic.test.ts‎

Lines changed: 8 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -1,34 +1,34 @@
11
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
22
import {
33
getDiagnosticSessionStateCountForTest,
4-
logSessionStateChange,
5-
resetDiagnosticStateForTest,
6-
} from "./diagnostic.js";
4+
getDiagnosticSessionState,
5+
resetDiagnosticSessionStateForTest,
6+
} from "./diagnostic-session-state.js";
77

88
describe("diagnostic session state pruning", () => {
99
beforeEach(() => {
1010
vi.useFakeTimers();
11-
resetDiagnosticStateForTest();
11+
resetDiagnosticSessionStateForTest();
1212
});
1313

1414
afterEach(() => {
15-
resetDiagnosticStateForTest();
15+
resetDiagnosticSessionStateForTest();
1616
vi.useRealTimers();
1717
});
1818

1919
it("evicts stale idle session states", () => {
20-
logSessionStateChange({ sessionId: "stale-1", state: "idle" });
20+
getDiagnosticSessionState({ sessionId: "stale-1" });
2121
expect(getDiagnosticSessionStateCountForTest()).toBe(1);
2222

2323
vi.advanceTimersByTime(31 * 60 * 1000);
24-
logSessionStateChange({ sessionId: "fresh-1", state: "idle" });
24+
getDiagnosticSessionState({ sessionId: "fresh-1" });
2525

2626
expect(getDiagnosticSessionStateCountForTest()).toBe(1);
2727
});
2828

2929
it("caps tracked session states to a bounded max", () => {
3030
for (let i = 0; i < 2001; i += 1) {
31-
logSessionStateChange({ sessionId: `session-${i}`, state: "idle" });
31+
getDiagnosticSessionState({ sessionId: `session-${i}` });
3232
}
3333

3434
expect(getDiagnosticSessionStateCountForTest()).toBe(2000);

‎src/logging/diagnostic.ts‎

Lines changed: 20 additions & 93 deletions
Original file line numberDiff line numberDiff line change
@@ -1,28 +1,17 @@
11
import { emitDiagnosticEvent } from "../infra/diagnostic-events.js";
2+
import {
3+
diagnosticSessionStates,
4+
getDiagnosticSessionState,
5+
getDiagnosticSessionStateCountForTest as getDiagnosticSessionStateCountForTestImpl,
6+
pruneDiagnosticSessionStates,
7+
resetDiagnosticSessionStateForTest,
8+
type SessionRef,
9+
type SessionStateValue,
10+
} from "./diagnostic-session-state.js";
211
import { createSubsystemLogger } from "./subsystem.js";
312

413
const diag = createSubsystemLogger("diagnostic");
514

6-
type SessionStateValue = "idle" | "processing" | "waiting";
7-
8-
type SessionState = {
9-
sessionId?: string;
10-
sessionKey?: string;
11-
lastActivity: number;
12-
state: SessionStateValue;
13-
queueDepth: number;
14-
};
15-
16-
type SessionRef = {
17-
sessionId?: string;
18-
sessionKey?: string;
19-
};
20-
21-
const sessionStates = new Map<string, SessionState>();
22-
const SESSION_STATE_TTL_MS = 30 * 60 * 1000;
23-
const SESSION_STATE_PRUNE_INTERVAL_MS = 60 * 1000;
24-
const SESSION_STATE_MAX_ENTRIES = 2000;
25-
2615
const webhookStats = {
2716
received: 0,
2817
processed: 0,
@@ -31,72 +20,11 @@ const webhookStats = {
3120
};
3221

3322
let lastActivityAt = 0;
34-
let lastSessionPruneAt = 0;
3523

3624
function markActivity() {
3725
lastActivityAt = Date.now();
3826
}
3927

40-
function pruneSessionStates(now = Date.now(), force = false): void {
41-
const shouldPruneForSize = sessionStates.size > SESSION_STATE_MAX_ENTRIES;
42-
if (!force && !shouldPruneForSize && now - lastSessionPruneAt < SESSION_STATE_PRUNE_INTERVAL_MS) {
43-
return;
44-
}
45-
lastSessionPruneAt = now;
46-
47-
for (const [key, state] of sessionStates.entries()) {
48-
const ageMs = now - state.lastActivity;
49-
const isIdle = state.state === "idle";
50-
if (isIdle && state.queueDepth <= 0 && ageMs > SESSION_STATE_TTL_MS) {
51-
sessionStates.delete(key);
52-
}
53-
}
54-
55-
if (sessionStates.size <= SESSION_STATE_MAX_ENTRIES) {
56-
return;
57-
}
58-
const excess = sessionStates.size - SESSION_STATE_MAX_ENTRIES;
59-
const ordered = Array.from(sessionStates.entries()).toSorted(
60-
(a, b) => a[1].lastActivity - b[1].lastActivity,
61-
);
62-
for (let i = 0; i < excess; i += 1) {
63-
const key = ordered[i]?.[0];
64-
if (!key) {
65-
break;
66-
}
67-
sessionStates.delete(key);
68-
}
69-
}
70-
71-
function resolveSessionKey({ sessionKey, sessionId }: SessionRef) {
72-
return sessionKey ?? sessionId ?? "unknown";
73-
}
74-
75-
function getSessionState(ref: SessionRef): SessionState {
76-
pruneSessionStates();
77-
const key = resolveSessionKey(ref);
78-
const existing = sessionStates.get(key);
79-
if (existing) {
80-
if (ref.sessionId) {
81-
existing.sessionId = ref.sessionId;
82-
}
83-
if (ref.sessionKey) {
84-
existing.sessionKey = ref.sessionKey;
85-
}
86-
return existing;
87-
}
88-
const created: SessionState = {
89-
sessionId: ref.sessionId,
90-
sessionKey: ref.sessionKey,
91-
lastActivity: Date.now(),
92-
state: "idle",
93-
queueDepth: 0,
94-
};
95-
sessionStates.set(key, created);
96-
pruneSessionStates(Date.now(), true);
97-
return created;
98-
}
99-
10028
export function logWebhookReceived(params: {
10129
channel: string;
10230
updateType?: string;
@@ -174,7 +102,7 @@ export function logMessageQueued(params: {
174102
channel?: string;
175103
source: string;
176104
}) {
177-
const state = getSessionState(params);
105+
const state = getDiagnosticSessionState(params);
178106
state.queueDepth += 1;
179107
state.lastActivity = Date.now();
180108
if (diag.isEnabled("debug")) {
@@ -244,7 +172,7 @@ export function logSessionStateChange(
244172
reason?: string;
245173
},
246174
) {
247-
const state = getSessionState(params);
175+
const state = getDiagnosticSessionState(params);
248176
const isProbeSession = state.sessionId?.startsWith("probe-") ?? false;
249177
const prevState = state.state;
250178
state.state = params.state;
@@ -274,7 +202,7 @@ export function logSessionStateChange(
274202
}
275203

276204
export function logSessionStuck(params: SessionRef & { state: SessionStateValue; ageMs: number }) {
277-
const state = getSessionState(params);
205+
const state = getDiagnosticSessionState(params);
278206
diag.warn(
279207
`stuck session: sessionId=${state.sessionId ?? "unknown"} sessionKey=${
280208
state.sessionKey ?? "unknown"
@@ -329,7 +257,7 @@ export function logRunAttempt(params: SessionRef & { runId: string; attempt: num
329257
}
330258

331259
export function logActiveRuns() {
332-
const activeSessions = Array.from(sessionStates.entries())
260+
const activeSessions = Array.from(diagnosticSessionStates.entries())
333261
.filter(([, s]) => s.state === "processing")
334262
.map(
335263
([id, s]) =>
@@ -347,14 +275,14 @@ export function startDiagnosticHeartbeat() {
347275
}
348276
heartbeatInterval = setInterval(() => {
349277
const now = Date.now();
350-
pruneSessionStates(now, true);
351-
const activeCount = Array.from(sessionStates.values()).filter(
278+
pruneDiagnosticSessionStates(now, true);
279+
const activeCount = Array.from(diagnosticSessionStates.values()).filter(
352280
(s) => s.state === "processing",
353281
).length;
354-
const waitingCount = Array.from(sessionStates.values()).filter(
282+
const waitingCount = Array.from(diagnosticSessionStates.values()).filter(
355283
(s) => s.state === "waiting",
356284
).length;
357-
const totalQueued = Array.from(sessionStates.values()).reduce(
285+
const totalQueued = Array.from(diagnosticSessionStates.values()).reduce(
358286
(sum, s) => sum + s.queueDepth,
359287
0,
360288
);
@@ -386,7 +314,7 @@ export function startDiagnosticHeartbeat() {
386314
queued: totalQueued,
387315
});
388316

389-
for (const [, state] of sessionStates) {
317+
for (const [, state] of diagnosticSessionStates) {
390318
const ageMs = now - state.lastActivity;
391319
if (state.state === "processing" && ageMs > 120_000) {
392320
logSessionStuck({
@@ -409,17 +337,16 @@ export function stopDiagnosticHeartbeat() {
409337
}
410338

411339
export function getDiagnosticSessionStateCountForTest(): number {
412-
return sessionStates.size;
340+
return getDiagnosticSessionStateCountForTestImpl();
413341
}
414342

415343
export function resetDiagnosticStateForTest(): void {
416-
sessionStates.clear();
344+
resetDiagnosticSessionStateForTest();
417345
webhookStats.received = 0;
418346
webhookStats.processed = 0;
419347
webhookStats.errors = 0;
420348
webhookStats.lastReceived = 0;
421349
lastActivityAt = 0;
422-
lastSessionPruneAt = 0;
423350
stopDiagnosticHeartbeat();
424351
}
425352

0 commit comments

Comments
 (0)