File size: 3,030 Bytes
76289e7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 | // Diagnostic runtime helpers expose process runtime facts for diagnostics.
import {
areDiagnosticsEnabledForProcess,
emitInternalDiagnosticEvent as emitDiagnosticEvent,
} from "../infra/diagnostic-events.js";
import { getDiagnosticSessionState, type SessionRef } from "./diagnostic-session-state.js";
import { createSubsystemLogger } from "./subsystem.js";
// Shared diagnostic logger and queue-activity event helpers.
const diag = createSubsystemLogger("diagnostic");
let lastActivityAt = 0;
/** Root diagnostic subsystem logger. */
export const diagnosticLogger = diag;
/** Marks that diagnostics emitted useful activity. */
export function markDiagnosticActivity(): void {
lastActivityAt = Date.now();
}
/** Returns the last diagnostic activity timestamp for watchdog-style checks. */
export function getLastDiagnosticActivityAt(): number {
return lastActivityAt;
}
/** Clears diagnostic activity state for tests. */
export function resetDiagnosticActivityForTest(): void {
lastActivityAt = 0;
}
type DiagnosticMessageQueueParams = SessionRef & {
channel?: string;
source: string;
};
/** Records queue activity while letting internal run owners distinguish steering from backlog. */
export function logMessageQueuedWithBacklogPolicy(
params: DiagnosticMessageQueueParams,
countsTowardBacklog: boolean,
): void {
if (!areDiagnosticsEnabledForProcess()) {
return;
}
const state = getDiagnosticSessionState(params);
if (countsTowardBacklog) {
state.queueDepth += 1;
}
state.lastActivity = Date.now();
state.generation = (state.generation ?? 0) + 1;
state.lastStuckWarnAgeMs = undefined;
state.lastLongRunningWarnAgeMs = undefined;
if (diag.isEnabled("debug")) {
diag.debug(
`message queued: sessionId=${state.sessionId ?? "unknown"} sessionKey=${
state.sessionKey ?? "unknown"
} source=${params.source} queueDepth=${state.queueDepth} sessionState=${state.state}`,
);
}
emitDiagnosticEvent({
type: "message.queued",
sessionId: state.sessionId,
sessionKey: state.sessionKey,
channel: params.channel,
source: params.source,
queueDepth: state.queueDepth,
});
markDiagnosticActivity();
}
/** Logs and emits a diagnostic event when work enters a serialized lane. */
export function logLaneEnqueue(lane: string, queueSize: number): void {
if (!areDiagnosticsEnabledForProcess()) {
return;
}
diag.debug(`lane enqueue: lane=${lane} queueSize=${queueSize}`);
emitDiagnosticEvent({
type: "queue.lane.enqueue",
lane,
queueSize,
});
markDiagnosticActivity();
}
/** Logs and emits a diagnostic event when work leaves a serialized lane. */
export function logLaneDequeue(lane: string, waitMs: number, queueSize: number): void {
if (!areDiagnosticsEnabledForProcess()) {
return;
}
diag.debug(`lane dequeue: lane=${lane} waitMs=${waitMs} queueSize=${queueSize}`);
emitDiagnosticEvent({
type: "queue.lane.dequeue",
lane,
queueSize,
waitMs,
});
markDiagnosticActivity();
}
|