mirror of
https://github.com/diegosouzapw/OmniRoute.git
synced 2026-08-23 23:52:18 +03:00
Compare commits
1 Commits
fix/11233-
...
fix/11244-
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
f92a9396df |
@@ -74,6 +74,7 @@ export async function GET(request: Request) {
|
||||
credentialHealthModule,
|
||||
localHealthModule,
|
||||
adaptiveAdmissionModule,
|
||||
chatAdmissionModule,
|
||||
settingsResult,
|
||||
connectionsResult,
|
||||
] = await Promise.allSettled([
|
||||
@@ -86,6 +87,7 @@ export async function GET(request: Request) {
|
||||
import("@/lib/credentialHealth/cache"),
|
||||
import("@/lib/localHealthCheck"),
|
||||
import("@omniroute/open-sse/services/admission/runtime.ts"),
|
||||
import("@/shared/middleware/chatBodyAdmission"),
|
||||
getCachedSettings(),
|
||||
getProviderConnections(),
|
||||
]);
|
||||
@@ -172,6 +174,17 @@ export async function GET(request: Request) {
|
||||
null
|
||||
)
|
||||
: null;
|
||||
// #11244: the STRUCTURAL admission gate (chatBodyAdmission.ts — bounded
|
||||
// heavyweight lease + shed counters), exposed next to but distinct from the
|
||||
// adaptive shadow-mode snapshot above. Additive key — nothing existing moves.
|
||||
const chatAdmission =
|
||||
chatAdmissionModule.status === "fulfilled"
|
||||
? readHealthValue(
|
||||
"chat admission",
|
||||
() => chatAdmissionModule.value.perConnectionAdmissionController.snapshot(),
|
||||
null
|
||||
)
|
||||
: null;
|
||||
|
||||
const payload = buildHealthPayload({
|
||||
appVersion: APP_CONFIG.version,
|
||||
@@ -200,6 +213,7 @@ export async function GET(request: Request) {
|
||||
activeSessionsByKey,
|
||||
credentialHealth,
|
||||
adaptiveAdmission,
|
||||
chatAdmission,
|
||||
});
|
||||
|
||||
healthPayloadCache = { payload, expiresAt: Date.now() + HEALTH_PAYLOAD_TTL_MS };
|
||||
@@ -218,6 +232,7 @@ export async function GET(request: Request) {
|
||||
quotaMonitor: { ...fallbackQuotaMonitorSummary, monitors: [] },
|
||||
sessions: { activeCount: 0, stickyBoundCount: 0, byApiKey: {}, top: [] },
|
||||
adaptiveAdmission: null,
|
||||
chatAdmission: null,
|
||||
dedup: { inflightRequests: 0 },
|
||||
});
|
||||
}
|
||||
|
||||
@@ -3,9 +3,48 @@ import {
|
||||
getCodexParentAccountDiagnostic,
|
||||
} from "@omniroute/open-sse/services/codexAccount/index.ts";
|
||||
import type { AdaptiveAdmissionPublicSnapshot } from "@omniroute/open-sse/services/admission/runtime.ts";
|
||||
import type { PerConnectionAdmissionController } from "@/shared/middleware/chatBodyAdmission";
|
||||
|
||||
type JsonRecord = Record<string, unknown>;
|
||||
|
||||
/** Process-wide structural chat-admission snapshot type (chatBodyAdmission.ts). */
|
||||
export type ChatAdmissionSnapshot = ReturnType<PerConnectionAdmissionController["snapshot"]>;
|
||||
|
||||
/**
|
||||
* Low-card structural chat-admission health summary (#11244) — the bounded
|
||||
* heavyweight-lease gate from chatBodyAdmission.ts (#10110/#10437), NOT the
|
||||
* adaptive shadow-mode layer above. Lane keys are opaque HMAC fairness
|
||||
* fingerprints (resolveSessionId), never raw credentials.
|
||||
*/
|
||||
export type ChatAdmissionHealthSummary = {
|
||||
activeHeavy: number;
|
||||
activeHealthyHeadroom: number;
|
||||
waiting: number;
|
||||
queuedBytes: number;
|
||||
shedTotal: number;
|
||||
shedsByReason: Record<string, number>;
|
||||
lanes: Array<{ key: string; waiting: number }>;
|
||||
};
|
||||
|
||||
/**
|
||||
* Explicit allowlisted projection of the structural admission snapshot.
|
||||
* Never spreads the snapshot — only the documented low-cardinality fields pass.
|
||||
*/
|
||||
export function projectChatAdmissionSummary(
|
||||
snapshot: ChatAdmissionSnapshot | null | undefined
|
||||
): ChatAdmissionHealthSummary | null {
|
||||
if (!snapshot || typeof snapshot !== "object") return null;
|
||||
return {
|
||||
activeHeavy: snapshot.activeHeavy,
|
||||
activeHealthyHeadroom: snapshot.activeHealthyHeadroom,
|
||||
waiting: snapshot.waiting,
|
||||
queuedBytes: snapshot.queuedBytes,
|
||||
shedTotal: snapshot.shedTotal,
|
||||
shedsByReason: { ...(snapshot.shedsByReason ?? {}) },
|
||||
lanes: (snapshot.lanes ?? []).map((lane) => ({ key: lane.key, waiting: lane.waiting })),
|
||||
};
|
||||
}
|
||||
|
||||
/** Low-card adaptive-admission health summary — no tenant/request/body/queue details. */
|
||||
export type AdaptiveAdmissionHealthSummary = {
|
||||
mode: AdaptiveAdmissionPublicSnapshot["mode"];
|
||||
@@ -160,6 +199,8 @@ interface BuildHealthPayloadOptions {
|
||||
};
|
||||
/** Optional injected public adaptive-admission snapshot; projected, never raw-spread. */
|
||||
adaptiveAdmission?: AdaptiveAdmissionPublicSnapshot | null;
|
||||
/** #11244: optional structural chat-admission snapshot; projected, never raw-spread. */
|
||||
chatAdmission?: ChatAdmissionSnapshot | null;
|
||||
}
|
||||
|
||||
function limitMonitors(monitors: QuotaMonitorSnapshot[], maxItems = 8): QuotaMonitorSnapshot[] {
|
||||
@@ -347,6 +388,7 @@ export function buildHealthPayload({
|
||||
activeSessionsByKey = {},
|
||||
credentialHealth,
|
||||
adaptiveAdmission = null,
|
||||
chatAdmission = null,
|
||||
buildSha = null,
|
||||
}: BuildHealthPayloadOptions) {
|
||||
const timestamp = new Date().toISOString();
|
||||
@@ -449,6 +491,9 @@ export function buildHealthPayload({
|
||||
sessions: buildSessionsSummary({ activeSessions, activeSessionsByKey }),
|
||||
credentialHealth, // may be undefined if credentialHealth module not loaded
|
||||
adaptiveAdmission: projectAdaptiveAdmissionSummary(adaptiveAdmission),
|
||||
// #11244: the STRUCTURAL gate (chatBodyAdmission.ts) next to the adaptive one —
|
||||
// distinct key so clients reading `adaptiveAdmission` are untouched.
|
||||
chatAdmission: projectChatAdmissionSummary(chatAdmission),
|
||||
dedup: {
|
||||
inflightRequests,
|
||||
},
|
||||
|
||||
@@ -16,6 +16,7 @@
|
||||
*/
|
||||
|
||||
import { CORS_HEADERS } from "../utils/cors";
|
||||
import { createLogger } from "../utils/logger";
|
||||
import { createHmac } from "crypto";
|
||||
import v8 from "node:v8";
|
||||
import { trackRequest } from "../../lib/gracefulShutdown";
|
||||
@@ -168,6 +169,47 @@ interface AdmissionWaiter {
|
||||
readonly resolve: () => void;
|
||||
}
|
||||
|
||||
/**
|
||||
* Why a structural shed (503 `chat_admission_busy`) happened (#11244):
|
||||
* - `queue_timeout`: the bounded wait expired with no heavyweight capacity freed
|
||||
* (includes the `queueMs=0` legacy immediate-reject path — capacity was busy at
|
||||
* the instant the request arrived).
|
||||
* - `queued_bytes_budget`: the queued-bytes heap valve (#9654 / U3) refused to
|
||||
* park the waiter because the buffered-body budget was already exhausted.
|
||||
*
|
||||
* A client abort mid-wait is deliberately NOT a shed: capacity was never denied,
|
||||
* the caller simply left (its 503 is dropped on the dead connection).
|
||||
*/
|
||||
export type ChatAdmissionShedReason = "queue_timeout" | "queued_bytes_budget";
|
||||
|
||||
/**
|
||||
* One structural-shed observation, emitted to the shed sink at warn level.
|
||||
* `lane` is the opaque fairness key — the HMAC fingerprint produced by
|
||||
* `resolveSessionId` (or "anonymous"/"default"), never a raw credential.
|
||||
*/
|
||||
export interface ChatAdmissionShedEvent {
|
||||
reason: ChatAdmissionShedReason;
|
||||
activeHeavy: number;
|
||||
waiting: number;
|
||||
queuedBytes: number;
|
||||
lane: string;
|
||||
}
|
||||
|
||||
export type ChatAdmissionShedSink = (event: ChatAdmissionShedEvent) => void;
|
||||
|
||||
const shedLog = createLogger("chat-admission");
|
||||
|
||||
/**
|
||||
* Default shed sink (#11244): exactly one structured warn per structural shed.
|
||||
* The 503 returns BEFORE request logging, so without this line a shed left no
|
||||
* trace anywhere. No raw credentials — `lane` is already the HMAC fingerprint,
|
||||
* and the shared logger's redaction hook (logRedaction.ts) is the safety net.
|
||||
* Nothing is logged for admitted requests (noise).
|
||||
*/
|
||||
function defaultChatAdmissionShedSink(event: ChatAdmissionShedEvent): void {
|
||||
shedLog.warn(event, "structural chat admission shed (chat_admission_busy)");
|
||||
}
|
||||
|
||||
/**
|
||||
* Process-local heavyweight reservation. The capacity check and increment execute in one
|
||||
* synchronous JavaScript turn, making acquisition atomic within an OmniRoute process.
|
||||
@@ -190,6 +232,12 @@ export class ChatAdmissionController {
|
||||
/** Keys in creation order; #fairCursor scans them round-robin. */
|
||||
#fairKeys: string[] = [];
|
||||
#fairCursor = 0;
|
||||
/** #11244: in-memory shed history (total + per reason). The 503 chat_admission_busy
|
||||
* response returns before request logging, so without these counters a structural
|
||||
* shed was invisible. Same in-memory lifetime as the rest of the snapshot state. */
|
||||
#shedTotal = 0;
|
||||
#shedsByReason = new Map<string, number>();
|
||||
readonly #onShed: ChatAdmissionShedSink;
|
||||
|
||||
constructor(
|
||||
readonly maxHeavyInFlight = 1,
|
||||
@@ -197,7 +245,10 @@ export class ChatAdmissionController {
|
||||
/** #10437: bounded extra capacity for the healthy-heap fast path. `0` disables
|
||||
* the bypass entirely — every busy request then falls through to the same
|
||||
* bounded-wait/shed path used under real heap pressure. */
|
||||
readonly healthyHeadroom = CHAT_ADMISSION_HEALTHY_HEADROOM
|
||||
readonly healthyHeadroom = CHAT_ADMISSION_HEALTHY_HEADROOM,
|
||||
/** #11244: sink notified once per structural shed. Defaults to the shared pino
|
||||
* logger (warn); tests inject a capture/no-op sink. */
|
||||
onShed: ChatAdmissionShedSink = defaultChatAdmissionShedSink
|
||||
) {
|
||||
if (!Number.isSafeInteger(maxHeavyInFlight) || maxHeavyInFlight < 1) {
|
||||
throw new RangeError("maxHeavyInFlight must be a positive integer");
|
||||
@@ -208,6 +259,7 @@ export class ChatAdmissionController {
|
||||
if (!Number.isSafeInteger(healthyHeadroom) || healthyHeadroom < 0) {
|
||||
throw new RangeError("healthyHeadroom must be a non-negative integer");
|
||||
}
|
||||
this.#onShed = onShed;
|
||||
}
|
||||
|
||||
get activeHeavy(): number {
|
||||
@@ -264,6 +316,37 @@ export class ChatAdmissionController {
|
||||
return out;
|
||||
}
|
||||
|
||||
/** Total structural sheds since process start (#11244). */
|
||||
get shedTotal(): number {
|
||||
return this.#shedTotal;
|
||||
}
|
||||
|
||||
/** Structural sheds by reason since process start (#11244). */
|
||||
get shedsByReason(): Record<string, number> {
|
||||
const out: Record<string, number> = {};
|
||||
for (const [reason, count] of this.#shedsByReason) out[reason] = count;
|
||||
return out;
|
||||
}
|
||||
|
||||
/**
|
||||
* Record one structural shed (503 chat_admission_busy) and notify the shed sink
|
||||
* (#11244). Called internally at every capacity-driven give-up point in
|
||||
* `acquireHeavyWithin`; public so the aggregate snapshot wiring and tests can
|
||||
* exercise the same single path. `lane` is the opaque fairness key (HMAC
|
||||
* fingerprint), never a raw credential.
|
||||
*/
|
||||
recordShed(reason: ChatAdmissionShedReason, lane = "default"): void {
|
||||
this.#shedTotal += 1;
|
||||
this.#shedsByReason.set(reason, (this.#shedsByReason.get(reason) ?? 0) + 1);
|
||||
this.#onShed({
|
||||
reason,
|
||||
activeHeavy: this.#activeHeavy,
|
||||
waiting: this.waitingCount,
|
||||
queuedBytes: this.#queuedBytes,
|
||||
lane,
|
||||
});
|
||||
}
|
||||
|
||||
tryAcquireHeavy(): ChatAdmissionLease | null {
|
||||
if (this.#activeHeavy >= this.maxHeavyInFlight) return null;
|
||||
this.#activeHeavy += 1;
|
||||
@@ -318,9 +401,16 @@ export class ChatAdmissionController {
|
||||
const lease = this.tryAcquireHeavy();
|
||||
if (lease) return lease;
|
||||
const remaining = deadline - Date.now();
|
||||
if (remaining <= 0) return null;
|
||||
if (remaining <= 0) {
|
||||
// Wait window exhausted (or queueMs=0 immediate reject) with capacity still
|
||||
// busy — the caller answers the retryable 503. Count it (#11244).
|
||||
this.recordShed("queue_timeout", sessionKey);
|
||||
return null;
|
||||
}
|
||||
// Heap valve: refuse to park when the queued-bytes budget is exhausted.
|
||||
if (queuedBytes > 0 && this.#queuedBytes + queuedBytes > this.maxQueuedBytes) {
|
||||
// Same retryable 503, distinct cause: the wait itself would amplify the heap.
|
||||
this.recordShed("queued_bytes_budget", sessionKey);
|
||||
return null;
|
||||
}
|
||||
this.#queuedBytes += queuedBytes;
|
||||
@@ -367,7 +457,13 @@ export class ChatAdmissionController {
|
||||
// Cancel the deadline timer when abort/release wins; a fired timer is a no-op.
|
||||
if (deadlineTimer) clearTimeout(deadlineTimer);
|
||||
if (onAbort) signal?.removeEventListener("abort", onAbort);
|
||||
if (timedOut) return null;
|
||||
if (timedOut) {
|
||||
// The deadline timer won the race: a genuine shed. When the client ABORT
|
||||
// won instead (signal aborted while parked), capacity was never denied —
|
||||
// the 503 is dropped on the dead connection, so it is not counted (#11244).
|
||||
if (!signal?.aborted) this.recordShed("queue_timeout", sessionKey);
|
||||
return null;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -483,11 +579,18 @@ export class PerConnectionAdmissionController {
|
||||
|
||||
constructor(
|
||||
readonly maxHeavyInFlight = 1,
|
||||
// Deprecated pre-#10110 lane-eviction knobs: accepted for API
|
||||
// compatibility and ignored — there are no per-session lanes to evict.
|
||||
_opts?: { maxSessions?: number; sessionTtlMs?: number }
|
||||
// `maxSessions`/`sessionTtlMs` are deprecated pre-#10110 lane-eviction knobs:
|
||||
// accepted for API compatibility and ignored — there are no per-session lanes
|
||||
// to evict. `onShed` (#11244) is live: it replaces the shed sink of the shared
|
||||
// controller (tests inject a capture/no-op sink; production keeps the pino warn).
|
||||
_opts?: { maxSessions?: number; sessionTtlMs?: number; onShed?: ChatAdmissionShedSink }
|
||||
) {
|
||||
this.#controller = new ChatAdmissionController(maxHeavyInFlight);
|
||||
this.#controller = new ChatAdmissionController(
|
||||
maxHeavyInFlight,
|
||||
undefined,
|
||||
undefined,
|
||||
_opts?.onShed
|
||||
);
|
||||
}
|
||||
|
||||
/** Returns the process-global budget — the same instance for every session. */
|
||||
@@ -497,20 +600,26 @@ export class PerConnectionAdmissionController {
|
||||
|
||||
/**
|
||||
* Process-wide aggregate snapshot for observability: global totals plus
|
||||
* per-key waiter depths. Keys are opaque scheduler keys, never raw
|
||||
* credentials.
|
||||
* per-key waiter depths and the #11244 shed history (total + per reason).
|
||||
* Keys are opaque scheduler keys, never raw credentials.
|
||||
*/
|
||||
snapshot(): {
|
||||
activeHeavy: number;
|
||||
activeHealthyHeadroom: number;
|
||||
queuedBytes: number;
|
||||
waiting: number;
|
||||
lanes: ReadonlyArray<{ key: string; waiting: number }>;
|
||||
shedTotal: number;
|
||||
shedsByReason: Record<string, number>;
|
||||
} {
|
||||
return {
|
||||
activeHeavy: this.#controller.activeHeavy,
|
||||
activeHealthyHeadroom: this.#controller.activeHealthyHeadroom,
|
||||
queuedBytes: this.#controller.queuedBytes,
|
||||
waiting: this.#controller.waitingCount,
|
||||
lanes: this.#controller.waitersByKey,
|
||||
shedTotal: this.#controller.shedTotal,
|
||||
shedsByReason: this.#controller.shedsByReason,
|
||||
};
|
||||
}
|
||||
|
||||
|
||||
238
tests/unit/chat-admission-visibility-11244.test.ts
Normal file
238
tests/unit/chat-admission-visibility-11244.test.ts
Normal file
@@ -0,0 +1,238 @@
|
||||
// #11244: visibility for the STRUCTURAL chat admission gate
|
||||
// (src/shared/middleware/chatBodyAdmission.ts — the bounded heavyweight lease +
|
||||
// healthy-headroom path from #10110/#10437, NOT the adaptive shadow-mode layer in
|
||||
// open-sse/services/admission/). The 503 chat_admission_busy shed returns BEFORE
|
||||
// request logging, so today a shed is invisible: no counter, no log line, and the
|
||||
// process-wide snapshot (PerConnectionAdmissionController.snapshot()) reports only
|
||||
// live state (activeHeavy/queuedBytes/waiting/lanes) with no shed history.
|
||||
//
|
||||
// These tests pin the observability contract WITHOUT changing admission behavior:
|
||||
// (a) every structural shed (503 chat_admission_busy) increments an in-memory
|
||||
// counter — total + per reason ("queue_timeout" when the bounded wait expires,
|
||||
// "queued_bytes_budget" when the queued-bytes heap valve refuses to park) —
|
||||
// while a client abort mid-wait is NOT a shed (capacity was never denied);
|
||||
// (b) the process-wide snapshot exposes shedTotal + shedsByReason next to the
|
||||
// existing live fields;
|
||||
// (c) each shed emits exactly one structured pino warn carrying
|
||||
// reason/activeHeavy/waiting and the HMAC session fingerprint — never the raw
|
||||
// API key (resolveSessionId already fingerprints the credential).
|
||||
import { test } from "node:test";
|
||||
import assert from "node:assert/strict";
|
||||
import { mkdtempSync, readFileSync, existsSync } from "node:fs";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
|
||||
// Configure the shared pino logger BEFORE importing the admission module — the
|
||||
// logger builds its transports at import time (see logger-redaction-wiring.test.ts
|
||||
// for the same pattern). JSON to a temp file keeps test (c)'s capture deterministic.
|
||||
const logDir = mkdtempSync(join(tmpdir(), "omniroute-admission-11244-"));
|
||||
const logFile = join(logDir, "app.log");
|
||||
process.env.NODE_ENV = "production";
|
||||
process.env.APP_LOG_TO_FILE = "true";
|
||||
process.env.APP_LOG_FILE_PATH = logFile;
|
||||
|
||||
const {
|
||||
ChatAdmissionController,
|
||||
PerConnectionAdmissionController,
|
||||
perConnectionAdmissionController,
|
||||
admitChatStructure,
|
||||
resolveSessionId,
|
||||
} = await import("../../src/shared/middleware/chatBodyAdmission.ts");
|
||||
|
||||
function heavyBody() {
|
||||
return {
|
||||
messages: Array.from({ length: 200 }, () => ({ role: "user", content: "x".repeat(40) })),
|
||||
tools: [] as unknown[],
|
||||
};
|
||||
}
|
||||
|
||||
const heapHealthy = () => false; // "not under pressure" — the healthy-heap fast path
|
||||
const heapPressured = () => true; // forces the bounded-wait/shed path deterministically
|
||||
const silentSink = () => {}; // keep non-logging tests off the pino transport
|
||||
|
||||
test("#11244 (a): a structural shed after the bounded wait increments shedTotal and shedsByReason", async () => {
|
||||
// Primary lease (1) + bounded healthy-headroom (1): two concurrent heavy requests
|
||||
// admit on a healthy heap; the third must wait queueMs and then shed with a 503.
|
||||
const controller = new ChatAdmissionController(1, undefined, 1, silentSink);
|
||||
|
||||
const first = await admitChatStructure(heavyBody(), null, {
|
||||
controller,
|
||||
heapPressureCheck: heapHealthy,
|
||||
queueMs: 0,
|
||||
});
|
||||
const second = await admitChatStructure(heavyBody(), null, {
|
||||
controller,
|
||||
heapPressureCheck: heapHealthy,
|
||||
queueMs: 0,
|
||||
});
|
||||
assert.equal(first.admit, true, "first heavy request takes the primary lease");
|
||||
assert.equal(second.admit, true, "second heavy request takes the bounded headroom lease");
|
||||
assert.equal(controller.shedTotal, 0, "admitted requests never count as sheds");
|
||||
assert.deepEqual(controller.shedsByReason, {});
|
||||
|
||||
const shed = await admitChatStructure(heavyBody(), null, {
|
||||
controller,
|
||||
heapPressureCheck: heapHealthy,
|
||||
queueMs: 50,
|
||||
});
|
||||
assert.equal(shed.admit, false, "third heavy request must shed once both budgets are busy");
|
||||
if (!shed.admit) {
|
||||
assert.equal(shed.response.status, 503);
|
||||
const payload = await shed.response.json();
|
||||
assert.equal(payload.error.code, "chat_admission_busy");
|
||||
}
|
||||
|
||||
assert.equal(
|
||||
controller.shedTotal,
|
||||
1,
|
||||
"the shed must be counted even though it skips request logging"
|
||||
);
|
||||
assert.deepEqual(
|
||||
controller.shedsByReason,
|
||||
{ queue_timeout: 1 },
|
||||
"a bounded wait that expires with no freed capacity is a queue_timeout shed"
|
||||
);
|
||||
|
||||
// Counters are history, not live state: releasing the leases must not rewind them.
|
||||
if (first.admit) first.lease?.release();
|
||||
if (second.admit) second.lease?.release();
|
||||
assert.equal(controller.shedTotal, 1, "shed history survives lease release");
|
||||
|
||||
// And a subsequently admitted request must not be counted.
|
||||
const fourth = await admitChatStructure(heavyBody(), null, {
|
||||
controller,
|
||||
heapPressureCheck: heapHealthy,
|
||||
queueMs: 0,
|
||||
});
|
||||
assert.equal(fourth.admit, true);
|
||||
assert.equal(controller.shedTotal, 1);
|
||||
if (fourth.admit) fourth.lease?.release();
|
||||
});
|
||||
|
||||
test("#11244 (a2): the queued-bytes heap valve rejection is counted with its own reason", async () => {
|
||||
// maxQueuedBytes smaller than the conservative 256KB structural wait weight: the
|
||||
// valve refuses to park and the shed must be distinguishable from a queue timeout.
|
||||
const controller = new ChatAdmissionController(1, 1024, 0, silentSink);
|
||||
|
||||
const first = await admitChatStructure(heavyBody(), null, {
|
||||
controller,
|
||||
heapPressureCheck: heapPressured,
|
||||
queueMs: 0,
|
||||
});
|
||||
assert.equal(first.admit, true);
|
||||
|
||||
const shed = await admitChatStructure(heavyBody(), null, {
|
||||
controller,
|
||||
heapPressureCheck: heapPressured,
|
||||
queueMs: 1000,
|
||||
});
|
||||
assert.equal(shed.admit, false);
|
||||
if (!shed.admit) assert.equal(shed.response.status, 503);
|
||||
assert.equal(controller.shedTotal, 1);
|
||||
assert.deepEqual(controller.shedsByReason, { queued_bytes_budget: 1 });
|
||||
|
||||
if (first.admit) first.lease?.release();
|
||||
});
|
||||
|
||||
test("#11244 (a3): a client abort while parked is not a shed — capacity was never denied", async () => {
|
||||
const controller = new ChatAdmissionController(1, undefined, 0, silentSink);
|
||||
|
||||
const first = await admitChatStructure(heavyBody(), null, {
|
||||
controller,
|
||||
heapPressureCheck: heapPressured,
|
||||
queueMs: 0,
|
||||
});
|
||||
assert.equal(first.admit, true);
|
||||
|
||||
const abort = new AbortController();
|
||||
const pending = admitChatStructure(heavyBody(), null, {
|
||||
controller,
|
||||
heapPressureCheck: heapPressured,
|
||||
queueMs: 5000,
|
||||
signal: abort.signal,
|
||||
});
|
||||
setTimeout(() => abort.abort(), 20);
|
||||
const result = await pending;
|
||||
assert.equal(
|
||||
result.admit,
|
||||
false,
|
||||
"the caller still answers a (dropped) 503 on the dead connection"
|
||||
);
|
||||
assert.equal(controller.shedTotal, 0, "an aborted wait frees capacity instead of shedding");
|
||||
assert.deepEqual(controller.shedsByReason, {});
|
||||
|
||||
if (first.admit) first.lease?.release();
|
||||
});
|
||||
|
||||
test("#11244 (b): the process-wide snapshot exposes shed counters next to the live fields", async () => {
|
||||
const empty = perConnectionAdmissionController.snapshot();
|
||||
assert.equal(typeof empty.activeHeavy, "number");
|
||||
assert.equal(typeof empty.queuedBytes, "number");
|
||||
assert.equal(typeof empty.waiting, "number");
|
||||
assert.ok(Array.isArray(empty.lanes));
|
||||
assert.equal(
|
||||
empty.shedTotal,
|
||||
0,
|
||||
"no shed happened through the production singleton in this process"
|
||||
);
|
||||
assert.deepEqual(empty.shedsByReason, {});
|
||||
|
||||
// A shed recorded through a session's controller surfaces in the aggregate snapshot.
|
||||
const pc = new PerConnectionAdmissionController(1, { onShed: silentSink });
|
||||
const controller = pc.getController("key_visibility11244");
|
||||
controller.recordShed("queue_timeout", "key_visibility11244");
|
||||
controller.recordShed("queue_timeout", "key_visibility11244");
|
||||
controller.recordShed("queued_bytes_budget", "key_visibility11244");
|
||||
|
||||
const snap = pc.snapshot();
|
||||
assert.equal(snap.shedTotal, 3);
|
||||
assert.deepEqual(snap.shedsByReason, { queue_timeout: 2, queued_bytes_budget: 1 });
|
||||
});
|
||||
|
||||
test("#11244 (c): each shed logs one structured warn with the session fingerprint, never the raw key", async () => {
|
||||
const rawKey = "visRAWSECRETtoken11244xyz"; // matches no logRedaction pattern — a leak would show verbatim
|
||||
const fingerprint = resolveSessionId(
|
||||
new Request("http://localhost/v1/chat/completions", {
|
||||
headers: { authorization: `Bearer ${rawKey}` },
|
||||
})
|
||||
);
|
||||
assert.ok(fingerprint.startsWith("key_"), "resolveSessionId returns the HMAC fingerprint");
|
||||
assert.ok(!fingerprint.includes(rawKey));
|
||||
|
||||
// Default sink (no injected onShed): the shed must go through the shared pino logger.
|
||||
const controller = new ChatAdmissionController(1);
|
||||
const primary = controller.tryAcquireHeavy();
|
||||
assert.ok(primary);
|
||||
|
||||
const shed = await admitChatStructure(heavyBody(), null, {
|
||||
controller,
|
||||
heapPressureCheck: heapPressured,
|
||||
queueMs: 25,
|
||||
sessionId: fingerprint,
|
||||
});
|
||||
assert.equal(shed.admit, false);
|
||||
primary.release();
|
||||
|
||||
// Poll the worker-thread-written log file until the shed line lands.
|
||||
const deadline = Date.now() + 4000;
|
||||
let contents = "";
|
||||
while (Date.now() < deadline) {
|
||||
if (existsSync(logFile)) {
|
||||
contents = readFileSync(logFile, "utf8");
|
||||
if (contents.includes("chat_admission_busy")) break;
|
||||
}
|
||||
await new Promise((resolve) => setTimeout(resolve, 50));
|
||||
}
|
||||
|
||||
assert.ok(contents.includes("chat_admission_busy"), "the shed log line names the rejection code");
|
||||
assert.match(
|
||||
contents,
|
||||
/"level":(40|"warn")/,
|
||||
"sheds log at warn level (numeric 40 when the file transport strips the level formatter)"
|
||||
);
|
||||
assert.ok(contents.includes('"module":"chat-admission"'), "the log is scoped to the gate");
|
||||
assert.ok(contents.includes('"reason":"queue_timeout"'), "the shed reason is structured");
|
||||
assert.ok(contents.includes('"activeHeavy":1'), "live state travels with the log line");
|
||||
assert.ok(contents.includes(fingerprint), "the lane fingerprint allows per-key correlation");
|
||||
assert.ok(!contents.includes(rawKey), "the raw API key must never reach the shed log");
|
||||
});
|
||||
@@ -6,6 +6,7 @@ import {
|
||||
buildSessionsSummary,
|
||||
buildTelemetryPayload,
|
||||
projectAdaptiveAdmissionSummary,
|
||||
projectChatAdmissionSummary,
|
||||
} from "../../src/lib/monitoring/observability.ts";
|
||||
|
||||
test("buildSessionsSummary returns sticky counts and ordered top sessions", () => {
|
||||
@@ -336,3 +337,71 @@ test("buildHealthPayload projects allowlisted adaptiveAdmission aggregates only"
|
||||
assert.equal(projectAdaptiveAdmissionSummary(null), null);
|
||||
assert.equal(projectAdaptiveAdmissionSummary(undefined), null);
|
||||
});
|
||||
|
||||
// #11244: the STRUCTURAL chat-admission gate (chatBodyAdmission.ts) must surface in
|
||||
// the health payload next to — never instead of — the adaptive snapshot, with only
|
||||
// the documented low-cardinality fields projected.
|
||||
test("buildHealthPayload projects allowlisted structural chatAdmission fields only", () => {
|
||||
const snapshot = {
|
||||
activeHeavy: 1,
|
||||
activeHealthyHeadroom: 1,
|
||||
waiting: 2,
|
||||
queuedBytes: 524_288,
|
||||
shedTotal: 3,
|
||||
shedsByReason: { queue_timeout: 2, queued_bytes_budget: 1 },
|
||||
lanes: [
|
||||
{ key: "key_c49d1c242feda590", waiting: 1 },
|
||||
{ key: "anonymous", waiting: 1 },
|
||||
],
|
||||
// Extra keys that must never leak into the public payload.
|
||||
internalController: { secret: "controller-state" },
|
||||
rawAuthorization: "Bearer raw-SHOULD-NOT-LEAK",
|
||||
} as unknown as import("../../src/lib/monitoring/observability.ts").ChatAdmissionSnapshot;
|
||||
|
||||
const payload = buildHealthPayload({
|
||||
appVersion: "9.9.9",
|
||||
settings: { setupComplete: false },
|
||||
connections: [],
|
||||
circuitBreakers: [],
|
||||
rateLimitStatus: {},
|
||||
learnedLimits: {},
|
||||
lockouts: {},
|
||||
localProviders: {},
|
||||
inflightRequests: 0,
|
||||
quotaMonitorSummary: {
|
||||
active: 0,
|
||||
alerting: 0,
|
||||
exhausted: 0,
|
||||
errors: 0,
|
||||
statusCounts: { starting: 0, idle: 0, healthy: 0, warning: 0, exhausted: 0, error: 0 },
|
||||
byProvider: {},
|
||||
},
|
||||
quotaMonitorMonitors: [],
|
||||
activeSessions: [],
|
||||
chatAdmission: snapshot,
|
||||
});
|
||||
|
||||
assert.deepEqual(payload.chatAdmission, {
|
||||
activeHeavy: 1,
|
||||
activeHealthyHeadroom: 1,
|
||||
waiting: 2,
|
||||
queuedBytes: 524_288,
|
||||
shedTotal: 3,
|
||||
shedsByReason: { queue_timeout: 2, queued_bytes_budget: 1 },
|
||||
lanes: [
|
||||
{ key: "key_c49d1c242feda590", waiting: 1 },
|
||||
{ key: "anonymous", waiting: 1 },
|
||||
],
|
||||
});
|
||||
// The adaptive projection is untouched by the new key.
|
||||
assert.equal(payload.adaptiveAdmission, null);
|
||||
|
||||
const json = JSON.stringify(payload);
|
||||
assert.equal(json.includes("controller-state"), false);
|
||||
assert.equal(json.includes("raw-SHOULD-NOT-LEAK"), false);
|
||||
assert.equal(json.includes("internalController"), false);
|
||||
|
||||
// Absent / null snapshot projects to null (degraded path parity).
|
||||
assert.equal(projectChatAdmissionSummary(null), null);
|
||||
assert.equal(projectChatAdmissionSummary(undefined), null);
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user