From dc9b2733fe22435939b10bc855beb430c267bd21 Mon Sep 17 00:00:00 2001 From: BigSimmo <87357024+BigSimmo@users.noreply.github.com> Date: Sat, 18 Jul 2026 03:26:45 +0800 Subject: [PATCH] feat: expose answer coalescing health metrics --- docs/capacity-review.md | 8 +++ docs/observability-slos.md | 22 +++++++-- src/app/api/health/ready/route.ts | 1 + src/lib/health-response.ts | 10 ++++ .../answer-coalescing-metrics.ts | 49 +++++++++++++++++++ src/lib/rag.ts | 15 +++++- tests/answer-coalescing-metrics.test.ts | 43 ++++++++++++++++ tests/health-route.test.ts | 12 ++++- 8 files changed, 154 insertions(+), 6 deletions(-) create mode 100644 src/lib/observability/answer-coalescing-metrics.ts create mode 100644 tests/answer-coalescing-metrics.test.ts diff --git a/docs/capacity-review.md b/docs/capacity-review.md index 6f23f5dd5..ce50f3da4 100644 --- a/docs/capacity-review.md +++ b/docs/capacity-review.md @@ -67,6 +67,14 @@ round, coalescing + cache absorb roughly a third of the fan-out at exactly the moment load is highest. This only works while the app is a **single process** (see deployment doc §2). +The authorized deep health probe now exposes privacy-safe process-local +`coalescing` counters (`originations`, `coalescedWaiters`, and +`activeOriginations`). Before changing replica count, compare their deltas with +the retrieval cache counters over a duplicate-heavy period. A sustained low +coalescing rate may be expected for unique queries, but is a capacity/cost +regression signal when the ward-round duplicate hypothesis holds; it is not a +reason to make readiness fail. + ### OpenAI: rate limits and generation concurrency Per answer: 1 embedding call (unless the lexical fast path skips it) + 1–2 diff --git a/docs/observability-slos.md b/docs/observability-slos.md index b1a3889fc..5b9790a8e 100644 --- a/docs/observability-slos.md +++ b/docs/observability-slos.md @@ -177,7 +177,7 @@ Operational notes: The §2 reliability SQL is now also a scrape. An **authorized deep probe** — `GET /api/health?deep=1` with the `x-health-deep-token: $HEALTH_DEEP_PROBE_SECRET` -header (same operator gate as the Supabase probe) — returns two counter blocks: +header (same operator gate as the Supabase probe) — returns three counter blocks: - **`slo`** — `answerSloSnapshot` (`src/lib/observability/answer-slo.ts`) counts `rag_queries` over the trailing `windowMinutes` (60) and reports @@ -199,12 +199,25 @@ header (same operator gate as the Supabase probe) — returns two counter blocks never write a telemetry row. `hitRate` is a convenience for eyeballing a single probe. +- **`coalescing`** — `answerCoalescingMetricsSnapshot` + (`src/lib/observability/answer-coalescing-metrics.ts`) reports cumulative + coalescible answer **originations**, duplicate **coalescedWaiters**, and the + current **activeOriginations** gauge for this one app process. It has no query, + owner, document, cache key, or clinical content. A scraper must derive a + windowed `coalescingRate` from counter deltas; the supplied rate is a + process-lifetime convenience. This is an operational capacity/cost signal, + not a liveness gate: a sustained near-zero rate during a known duplicate-heavy + ward round means cache keys, cache settings, or replica dilution should be + investigated before adding app replicas. + ```jsonc // GET /api/health?deep=1 (authorized, live) "slo": { "windowMinutes": 60, "totalQueries": 412, "hybridRpcErrorQueries": 0, "hybridRpcErrorRate": 0.0, "degradedQueries": 26, "degradedRate": 0.063 }, -"cache": { "lookups": 1873, "hits": 1402, "misses": 471, "hitRate": 0.748 } +"cache": { "lookups": 1873, "hits": 1402, "misses": 471, "hitRate": 0.748 }, +"coalescing": { "originations": 312, "coalescedWaiters": 100, + "activeOriginations": 4, "coalescingRate": 0.243 } ``` **The counter _values_ never flip liveness.** A bad SLO rate or a cold cache @@ -214,7 +227,10 @@ only by the reachability `checks`: a failing `probeSupabaseHealth()` still sets `checks.supabase = "error"` and returns `503`, whereas an `answerSloSnapshot()` query failure merely omits the `slo` block (never a false-healthy zero). The cache counter is always present for a token-authorized probe (in-process, works -in demo mode). Both blocks are wired through `src/lib/health-response.ts` and are +in demo mode). The cache and coalescing blocks are process-local and reset on +deploy/restart, so operators must use per-process deltas rather than combine raw +counter values across replicas. All three blocks are wired through +`src/lib/health-response.ts` and are **withheld from the unauthenticated `/api/health/ready` endpoint** (Railway's readiness target, which exposes no diagnostic details). The §2 warn/page thresholds map directly onto these fields. diff --git a/src/app/api/health/ready/route.ts b/src/app/api/health/ready/route.ts index 9e7c6a463..92c54aff3 100644 --- a/src/app/api/health/ready/route.ts +++ b/src/app/api/health/ready/route.ts @@ -11,5 +11,6 @@ export async function GET(request: Request) { allowUnauthenticatedDeep: true, includeSlo: false, includeCache: false, + includeCoalescing: false, }); } diff --git a/src/lib/health-response.ts b/src/lib/health-response.ts index 848aad71d..42dd08266 100644 --- a/src/lib/health-response.ts +++ b/src/lib/health-response.ts @@ -3,6 +3,10 @@ import { allowDeepHealthProbe } from "@/lib/deep-probe-auth"; import { env, isDemoMode } from "@/lib/env"; import type { AnswerSloSnapshot, SloProbeClient } from "@/lib/observability/answer-slo"; import { cacheMetricsSnapshot, type CacheMetricsSnapshot } from "@/lib/observability/cache-metrics"; +import { + answerCoalescingMetricsSnapshot, + type AnswerCoalescingMetricsSnapshot, +} from "@/lib/observability/answer-coalescing-metrics"; import type { SpendProbeClient, SpendSnapshot } from "@/lib/observability/spend-metrics"; type HealthResponseOptions = { @@ -10,6 +14,7 @@ type HealthResponseOptions = { allowUnauthenticatedDeep?: boolean; includeSlo?: boolean; includeCache?: boolean; + includeCoalescing?: boolean; includeSpend?: boolean; }; @@ -23,6 +28,7 @@ export async function healthResponse(request: Request, options: HealthResponseOp }; let slo: AnswerSloSnapshot | null = null; let cache: CacheMetricsSnapshot | null = null; + let coalescing: AnswerCoalescingMetricsSnapshot | null = null; let spend: SpendSnapshot | null = null; if (deep) { @@ -39,6 +45,9 @@ export async function healthResponse(request: Request, options: HealthResponseOp if (tokenAuthorized && options.includeCache !== false) { cache = cacheMetricsSnapshot(); } + if (tokenAuthorized && options.includeCoalescing !== false) { + coalescing = answerCoalescingMetricsSnapshot(); + } if (supabaseConfigured && !isDemoMode()) { try { @@ -103,6 +112,7 @@ export async function healthResponse(request: Request, options: HealthResponseOp checks, ...(slo ? { slo } : {}), ...(cache ? { cache } : {}), + ...(coalescing ? { coalescing } : {}), ...(spend ? { spend } : {}), }, { status: ready ? 200 : 503, headers: { "Cache-Control": "no-store" } }, diff --git a/src/lib/observability/answer-coalescing-metrics.ts b/src/lib/observability/answer-coalescing-metrics.ts new file mode 100644 index 000000000..32161ee13 --- /dev/null +++ b/src/lib/observability/answer-coalescing-metrics.ts @@ -0,0 +1,49 @@ +// Process-local counters for the in-flight answer coalescer. +// +// These measurements intentionally contain no request key, owner, query, or +// document data. They show whether this process is absorbing duplicate answer +// work, which is otherwise invisible when considering replica changes. + +let originations = 0; +let coalescedWaiters = 0; +let activeOriginations = 0; + +export type AnswerCoalescingMetricsSnapshot = { + originations: number; + coalescedWaiters: number; + activeOriginations: number; + // 0..1; 0 when no coalescible request has completed or joined yet. + coalescingRate: number; +}; + +export function recordAnswerOrigination(): void { + originations += 1; + activeOriginations += 1; +} + +export function recordAnswerOriginationFinished(): void { + // A defensive floor means an unexpected cleanup path cannot emit a negative + // gauge, which would be misleading to the operator polling this endpoint. + activeOriginations = Math.max(0, activeOriginations - 1); +} + +export function recordCoalescedAnswerWaiter(): void { + coalescedWaiters += 1; +} + +export function answerCoalescingMetricsSnapshot(): AnswerCoalescingMetricsSnapshot { + const totalRequests = originations + coalescedWaiters; + return { + originations, + coalescedWaiters, + activeOriginations, + coalescingRate: totalRequests > 0 ? coalescedWaiters / totalRequests : 0, + }; +} + +/** Test-only: reset process-local counters between cases. */ +export function resetAnswerCoalescingMetrics(): void { + originations = 0; + coalescedWaiters = 0; + activeOriginations = 0; +} diff --git a/src/lib/rag.ts b/src/lib/rag.ts index a2942d577..f5d43bb3f 100644 --- a/src/lib/rag.ts +++ b/src/lib/rag.ts @@ -119,6 +119,11 @@ export { retrievalPlanCacheQuery, } from "@/lib/rag-cache"; import { classifySearchCacheOutcome, recordCacheLookup } from "@/lib/observability/cache-metrics"; +import { + recordAnswerOrigination, + recordAnswerOriginationFinished, + recordCoalescedAnswerWaiter, +} from "@/lib/observability/answer-coalescing-metrics"; import { buildRagSourceBlock, compactContextText, neutralizeIdentityField } from "@/lib/rag-source-block"; export { buildRagSourceBlock, truncateForModel } from "@/lib/rag-source-block"; import { @@ -3118,6 +3123,7 @@ export async function answerQuestionWithScope(args: AnswerQuestionWithScopeArgs) let existing = inflightKey ? answerInflight.get(inflightKey) : undefined; while (existing) { + recordCoalescedAnswerWaiter(); await args.onProgress?.({ stage: "cached", message: "Waiting for an identical cited answer request already in progress.", @@ -3149,8 +3155,15 @@ export async function answerQuestionWithScope(args: AnswerQuestionWithScopeArgs) } } + // Only coalescible requests belong in this process-local signal. Requests + // that intentionally bypass cache/coalescing must not make a replica look + // ineffective, and neither keys nor clinical content leave this function. + if (inflightKey) recordAnswerOrigination(); const pending = answerQuestionWithScopeUncoalesced(args, startedAt).finally(() => { - if (inflightKey) answerInflight.delete(inflightKey); + if (inflightKey) { + answerInflight.delete(inflightKey); + recordAnswerOriginationFinished(); + } }); if (inflightKey) answerInflight.set(inflightKey, pending); return pending; diff --git a/tests/answer-coalescing-metrics.test.ts b/tests/answer-coalescing-metrics.test.ts new file mode 100644 index 000000000..d50eb7320 --- /dev/null +++ b/tests/answer-coalescing-metrics.test.ts @@ -0,0 +1,43 @@ +import { afterEach, describe, expect, it } from "vitest"; + +import { + answerCoalescingMetricsSnapshot, + recordAnswerOrigination, + recordAnswerOriginationFinished, + recordCoalescedAnswerWaiter, + resetAnswerCoalescingMetrics, +} from "@/lib/observability/answer-coalescing-metrics"; + +afterEach(() => { + resetAnswerCoalescingMetrics(); +}); + +describe("answer coalescing metrics", () => { + it("reports a safe zero snapshot before coalescible answer work", () => { + expect(answerCoalescingMetricsSnapshot()).toEqual({ + originations: 0, + coalescedWaiters: 0, + activeOriginations: 0, + coalescingRate: 0, + }); + }); + + it("tracks originations, waiters, and the active gauge without request content", () => { + recordAnswerOrigination(); + recordAnswerOrigination(); + recordCoalescedAnswerWaiter(); + recordAnswerOriginationFinished(); + + expect(answerCoalescingMetricsSnapshot()).toEqual({ + originations: 2, + coalescedWaiters: 1, + activeOriginations: 1, + coalescingRate: 1 / 3, + }); + }); + + it("never reports a negative active gauge when cleanup is repeated", () => { + recordAnswerOriginationFinished(); + expect(answerCoalescingMetricsSnapshot().activeOriginations).toBe(0); + }); +}); diff --git a/tests/health-route.test.ts b/tests/health-route.test.ts index c5b90e44e..0a62475c6 100644 --- a/tests/health-route.test.ts +++ b/tests/health-route.test.ts @@ -85,7 +85,7 @@ describe("GET /api/health", () => { expect(body.checks).toMatchObject({ supabaseConfig: "ok", openaiConfig: "skipped" }); }); - it("gates the deep probe without a token and omits the slo/cache snapshots", async () => { + it("gates the deep probe without a token and omits diagnostic snapshots", async () => { mockEnv({ configured: true, deepSecret: true }); const { GET } = await import("../src/app/api/health/route"); @@ -96,6 +96,7 @@ describe("GET /api/health", () => { expect(body.checks).toMatchObject({ supabase: "unauthorized" }); expect(body.slo).toBeUndefined(); expect(body.cache).toBeUndefined(); + expect(body.coalescing).toBeUndefined(); }); it("exposes the in-process cache hit-rate counter on an authorized deep probe", async () => { @@ -115,6 +116,12 @@ describe("GET /api/health", () => { expect(cache.misses).toBe(cache.lookups - cache.hits); expect(cache.hitRate).toBeGreaterThanOrEqual(0); expect(cache.hitRate).toBeLessThanOrEqual(1); + const coalescing = body.coalescing as Record; + expect(typeof coalescing.originations).toBe("number"); + expect(typeof coalescing.coalescedWaiters).toBe("number"); + expect(typeof coalescing.activeOriginations).toBe("number"); + expect(coalescing.coalescingRate).toBeGreaterThanOrEqual(0); + expect(coalescing.coalescingRate).toBeLessThanOrEqual(1); expect(body.slo).toBeUndefined(); }); @@ -142,7 +149,7 @@ describe("GET /api/health/ready", () => { expect(body.checks).toMatchObject({ supabaseConfig: "ok", openaiConfig: "ok", supabase: "skipped" }); }); - it("exposes no diagnostic details (slo/cache) even to a token-bearing caller", async () => { + it("exposes no diagnostic details even to a token-bearing caller", async () => { mockEnv({ configured: true, demoMode: true, deepSecret: true }); const { GET } = await import("../src/app/api/health/ready/route"); @@ -153,5 +160,6 @@ describe("GET /api/health/ready", () => { expect(body.slo).toBeUndefined(); expect(body.cache).toBeUndefined(); + expect(body.coalescing).toBeUndefined(); }); });