From c02ebce8514247316e388bf3bd2120394eec83fd Mon Sep 17 00:00:00 2001 From: William Callahan Date: Sun, 1 Nov 2026 16:02:44 -0800 Subject: [PATCH 1/9] fix(search): retain correlated request and failure diagnostics Correlate HTTP and SSE failures and retain browser observations in the existing application logs. Preserve provider 429 causes and retry headers without replaying searches or changing rate limits. Agent-Session: ses_e6699fbbdffeNhL8YBrQk84Urg --- docs/api.md | 18 +++ frontend/e2e/search-diagnostics.spec.ts | 45 +++++++ frontend/src/lib/services/csrf.ts | 8 +- .../lib/services/searchDiagnostics.test.ts | 110 ++++++++++++++++++ .../src/lib/services/searchDiagnostics.ts | 33 ++++++ frontend/src/lib/services/sse.test.ts | 37 ++++++ frontend/src/lib/services/sse.ts | 80 +++++++++++-- frontend/src/lib/validation/schemas.ts | 17 +++ .../in/web/ApiRequestLoggingFilter.java | 80 +++++++++++++ .../web/BrowserSearchFailureController.java | 84 +++++++++++++ .../OpenAiStreamingFailureException.java | 21 +++- .../OpenAiCompatibleEmbeddingClient.java | 50 ++++++-- .../javachat/web/ChatController.java | 105 +++++++++-------- .../web/GuidedLearningController.java | 58 +++++---- .../javachat/web/SseSupport.java | 33 +++++- .../BrowserSearchFailureControllerTest.java | 80 +++++++++++++ .../OpenAiStreamingFailureExceptionTest.java | 4 + .../javachat/config/SecurityConfigTest.java | 55 +++++++++ .../OpenAiCompatibleEmbeddingClientTest.java | 65 +++++++++++ .../javachat/web/SseSupportTest.java | 31 +++++ 20 files changed, 919 insertions(+), 95 deletions(-) create mode 100644 frontend/e2e/search-diagnostics.spec.ts create mode 100644 frontend/src/lib/services/searchDiagnostics.test.ts create mode 100644 frontend/src/lib/services/searchDiagnostics.ts create mode 100644 src/main/java/com/williamcallahan/javachat/adapters/in/web/ApiRequestLoggingFilter.java create mode 100644 src/main/java/com/williamcallahan/javachat/adapters/in/web/BrowserSearchFailureController.java create mode 100644 src/test/java/com/williamcallahan/javachat/adapters/in/web/BrowserSearchFailureControllerTest.java diff --git a/docs/api.md b/docs/api.md index 19a527ec..7eb5536a 100644 --- a/docs/api.md +++ b/docs/api.md @@ -50,6 +50,24 @@ arrive before the `provider` event. Chat uses the shared OpenAI-compatible gateway with Claude Haiku 5.5 at low reasoning effort. An unavailable, rate-limited, or failed gateway request terminates the stream with an `error` event; no alternate provider exists. +API responses include `X-Request-ID`, including security rejections and HTTP-200 SSE streams. +Application logs distinguish HTTP completion (`api_request_completed`) from a terminal SSE +error (`search_stream_failed`); HTTP 200 alone does not establish that the answer succeeded. +Embedding rate-limit records retain the upstream status, workload, model, gateway request ID, +and `Retry-After` when supplied. Missing retry timing does not replace the original provider cause. + +### Browser failure diagnostics + +POST `/api/diagnostics/search-failures` accepts bounded search-transport observations and returns +202. It requires the same CSRF protection as other browser POSTs. Reports identify the API path, +failure kind (`HTTP`, `SSE`, `NETWORK`, or `PROTOCOL`), optional HTTP status/request ID/error code/ +retry timing, page path, loaded module asset, and elapsed milliseconds. URL queries, prompts, +cookies, authorization headers, and raw response bodies are not report fields and are not emitted into logs. + +Browser reports use `event=browser_search_failed` and `provenance=client_reported`; they are not +verified server failures. A search failure sends one report without replaying the search. A failed +diagnostic submission remains console-visible and never recursively reports itself. + Example: ```bash diff --git a/frontend/e2e/search-diagnostics.spec.ts b/frontend/e2e/search-diagnostics.spec.ts new file mode 100644 index 00000000..8b234ed6 --- /dev/null +++ b/frontend/e2e/search-diagnostics.spec.ts @@ -0,0 +1,45 @@ +import { expect, test } from "@playwright/test"; +import { BrowserSearchFailureSchema } from "../src/lib/validation/schemas"; + +for (const failureKind of ["HTTP", "SSE"] as const) { + test(`${failureKind} search failure reaches diagnostics without replaying the question`, async ({ + page, + }) => { + const searchRequests: string[] = []; + await page.route("**/api/security/csrf", async (route) => route.fulfill({ status: 204 })); + await page.route("**/api/diagnostics/search-failures", async (route) => + route.fulfill({ status: 202 }), + ); + await page.route("**/api/chat/stream", async (route) => { + searchRequests.push(route.request().url()); + await route.fulfill({ + status: failureKind === "HTTP" ? 429 : 200, + headers: { "X-Request-ID": "browser-refusal-test", "Retry-After": "12" }, + contentType: failureKind === "HTTP" ? "application/json" : "text/event-stream", + body: + failureKind === "HTTP" + ? JSON.stringify({ status: "error", message: "Search is temporarily rate limited" }) + : 'event:error\ndata:{"message":"Search is temporarily rate limited","code":"test.rate_limited","retryable":true}\n\n', + }); + }); + await page.goto("/"); + const diagnosticSubmission = page.waitForRequest("**/api/diagnostics/search-failures"); + await page.getByLabel("Message input").fill("Explain Java records"); + await page.getByLabel("Message input").press("Enter"); + await expect( + page.getByText("Search is temporarily rate limited", { exact: true }), + ).toBeVisible(); + const diagnosticRequest = await diagnosticSubmission; + const parsedDiagnostic: unknown = diagnosticRequest.postDataJSON(); + const diagnostic = BrowserSearchFailureSchema.safeParse(parsedDiagnostic); + if (!diagnostic.success) throw diagnostic.error; + expect(diagnostic.data).toMatchObject({ + path: "/api/chat/stream", + kind: failureKind, + httpStatus: failureKind === "HTTP" ? 429 : 200, + requestId: "browser-refusal-test", + }); + expect(diagnosticRequest.postData()).not.toContain("Explain Java records"); + expect(searchRequests).toHaveLength(1); + }); +} diff --git a/frontend/src/lib/services/csrf.ts b/frontend/src/lib/services/csrf.ts index 22506f7f..f932083f 100644 --- a/frontend/src/lib/services/csrf.ts +++ b/frontend/src/lib/services/csrf.ts @@ -52,7 +52,7 @@ export function csrfHeader(): Record { return { [CSRF_HEADER_NAME]: tokenText }; } -async function readCsrfError( +export async function readApiError( httpResponse: Response, source: string, ): Promise { @@ -84,7 +84,7 @@ export async function extractApiErrorMessage( httpResponse: Response, source: string, ): Promise { - const apiError = await readCsrfError(httpResponse, source); + const apiError = await readApiError(httpResponse, source); if (!apiError) { return null; } @@ -154,7 +154,7 @@ export async function fetchWithCsrfRetry( return initialHttpResponse; } - const initialCsrfError = await readCsrfError(initialHttpResponse.clone(), source); + const initialCsrfError = await readApiError(initialHttpResponse.clone(), source); if (!initialCsrfError || !isRecoverableCsrfErrorMessage(initialCsrfError.message)) { return initialHttpResponse; } @@ -174,7 +174,7 @@ export async function fetchWithCsrfRetry( return retriedHttpResponse; } - const retriedCsrfError = await readCsrfError(retriedHttpResponse.clone(), `${source}:retry`); + const retriedCsrfError = await readApiError(retriedHttpResponse.clone(), `${source}:retry`); if (retriedCsrfError && isRecoverableCsrfErrorMessage(retriedCsrfError.message)) { maybeToastExpired(); } diff --git a/frontend/src/lib/services/searchDiagnostics.test.ts b/frontend/src/lib/services/searchDiagnostics.test.ts new file mode 100644 index 00000000..86824229 --- /dev/null +++ b/frontend/src/lib/services/searchDiagnostics.test.ts @@ -0,0 +1,110 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import { BrowserSearchFailureSchema } from "../validation/schemas"; +import { streamSse } from "./sse"; + +describe("search failure diagnostic transport", () => { + afterEach(() => { + vi.unstubAllGlobals(); + vi.restoreAllMocks(); + }); + + it("retains one HTTP refusal report without replaying the search or transmitting its prompt", async () => { + const submittedReports: string[] = []; + const searchRequests: string[] = []; + vi.stubGlobal( + "fetch", + vi.fn(async (url: string, options?: RequestInit) => { + if (url === "/api/diagnostics/search-failures") { + if (typeof options?.body !== "string") + throw new Error("Expected a JSON diagnostic request body"); + submittedReports.push(options.body); + return new Response(null, { status: 202 }); + } + searchRequests.push(url); + return new Response(null, { + status: 429, + headers: { + "X-Request-ID": "refusal-report-test", + "Retry-After": "12", + }, + }); + }), + ); + + await expect( + streamSse( + "/api/chat/stream", + { latest: "private search question" }, + { onText: vi.fn() }, + "chat.ts", + ), + ).rejects.toMatchObject({ httpStatus: 429 }); + await vi.waitFor(() => expect(submittedReports).toHaveLength(1)); + expect(searchRequests).toEqual(["/api/chat/stream"]); + expect(submittedReports[0]).not.toContain("private search question"); + const diagnostic = BrowserSearchFailureSchema.safeParse( + JSON.parse(submittedReports[0] ?? "null"), + ); + if (!diagnostic.success) throw diagnostic.error; + expect(diagnostic.data).toMatchObject({ + kind: "HTTP", + path: "/api/chat/stream", + httpStatus: 429, + requestId: "refusal-report-test", + retryAfterSeconds: 12, + }); + }); + + it("distinguishes an SSE error inside HTTP 200 from an HTTP refusal", async () => { + const submittedReports: string[] = []; + vi.stubGlobal( + "fetch", + vi.fn(async (url: string, options?: RequestInit) => { + if (url === "/api/diagnostics/search-failures") { + if (typeof options?.body !== "string") + throw new Error("Expected a JSON diagnostic request body"); + submittedReports.push(options.body); + return new Response(null, { status: 202 }); + } + return new Response( + 'event:error\ndata:{"message":"Temporarily unavailable","code":"stream.provider_retryable_error","retryable":true}\n\n', + { status: 200, headers: { "X-Request-ID": "sse-report-test" } }, + ); + }), + ); + await expect( + streamSse( + "/api/guided/stream", + { latest: "private question" }, + { onText: vi.fn() }, + "guided.ts", + ), + ).rejects.toThrow("Temporarily unavailable"); + await vi.waitFor(() => expect(submittedReports).toHaveLength(1)); + expect(JSON.parse(submittedReports[0] ?? "null")).toMatchObject({ + kind: "SSE", + httpStatus: 200, + requestId: "sse-report-test", + code: "stream.provider_retryable_error", + }); + }); + + it("does not recursively report an unavailable diagnostic endpoint", async () => { + const requestedPaths: string[] = []; + vi.stubGlobal( + "fetch", + vi.fn(async (url: string) => { + requestedPaths.push(url); + throw new TypeError("Connection unavailable"); + }), + ); + vi.spyOn(console, "error").mockImplementation(() => undefined); + vi.spyOn(console, "warn").mockImplementation(() => undefined); + await expect( + streamSse("/api/chat/stream", { latest: "question" }, { onText: vi.fn() }, "chat.ts"), + ).rejects.toThrow("Couldn't reach the server"); + await vi.waitFor(() => + expect(requestedPaths).toEqual(["/api/chat/stream", "/api/diagnostics/search-failures"]), + ); + }); +}); diff --git a/frontend/src/lib/services/searchDiagnostics.ts b/frontend/src/lib/services/searchDiagnostics.ts new file mode 100644 index 00000000..0098d3b9 --- /dev/null +++ b/frontend/src/lib/services/searchDiagnostics.ts @@ -0,0 +1,33 @@ +import { BrowserSearchFailureSchema, type BrowserSearchFailure } from "../validation/schemas"; +import { validateWithSchema } from "../validation/validate"; +import { fetchWithCsrfRetry } from "./csrf"; + +/** Sends one browser observation into the existing application log transport, without replaying searches. */ +export async function reportSearchFailure(failure: BrowserSearchFailure): Promise { + const diagnosticValidation = validateWithSchema( + BrowserSearchFailureSchema, + failure, + "search-diagnostics", + ); + if (!diagnosticValidation.success) { + return; + } + try { + const diagnosticReply = await fetchWithCsrfRetry( + "/api/diagnostics/search-failures", + { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify(diagnosticValidation.validated), + keepalive: true, + }, + "search-diagnostics", + ); + if (!diagnosticReply.ok) { + console.warn("Search diagnostics were rejected", { status: diagnosticReply.status }); + } + } catch (diagnosticFailure) { + // A failed report must neither replace the search's original error nor recursively report itself. + console.warn("Search diagnostics could not reach the server", diagnosticFailure); + } +} diff --git a/frontend/src/lib/services/sse.test.ts b/frontend/src/lib/services/sse.test.ts index 00822ef7..839706b5 100644 --- a/frontend/src/lib/services/sse.test.ts +++ b/frontend/src/lib/services/sse.test.ts @@ -1,6 +1,10 @@ import { afterEach, describe, expect, it, vi } from "vitest"; import { streamSse, streamSseGet } from "./sse"; +vi.mock("./searchDiagnostics", () => ({ + reportSearchFailure: vi.fn().mockResolvedValue(undefined), +})); + const SSE_STREAM_RESPONSE_STATUS = 200; const HTTP_BAD_REQUEST_STATUS = 400; const HTTP_TOO_MANY_REQUESTS_STATUS = 429; @@ -180,6 +184,39 @@ describe("streamSse transport handling", () => { }); }); + it("preserves HTTP refusal diagnostics and authoritative retry timing", async () => { + vi.stubGlobal( + "fetch", + vi.fn().mockResolvedValue( + new Response( + JSON.stringify({ + status: "error", + message: "Search is temporarily rate limited", + details: "The upstream request budget resets shortly", + }), + { + status: HTTP_TOO_MANY_REQUESTS_STATUS, + headers: { + "Content-Type": "application/json", + "X-Request-ID": "search-rejection-test", + "Retry-After": "12", + }, + }, + ), + ), + ); + + await expect( + streamSseGet("/api/chat/stream", { onText: vi.fn() }, "sse.test.ts"), + ).rejects.toMatchObject({ + message: "Search is temporarily rate limited", + details: "The upstream request budget resets shortly", + httpStatus: HTTP_TOO_MANY_REQUESTS_STATUS, + requestId: "search-rejection-test", + retryAfterSeconds: 12, + }); + }); + it("reports a 4xx client error as a non-retryable stream failure", async () => { const badRequestMessage = `HTTP ${HTTP_BAD_REQUEST_STATUS}: Bad Request`; vi.stubGlobal( diff --git a/frontend/src/lib/services/sse.ts b/frontend/src/lib/services/sse.ts index cf75714c..2e8bb58c 100644 --- a/frontend/src/lib/services/sse.ts +++ b/frontend/src/lib/services/sse.ts @@ -16,9 +16,11 @@ import { type StreamError, type ProviderEvent, type Citation, + type BrowserSearchFailure, } from "../validation/schemas"; import { validateWithSchema } from "../validation/validate"; -import { csrfHeader, extractApiErrorMessage, fetchWithCsrfRetry } from "./csrf"; +import { csrfHeader, readApiError, fetchWithCsrfRetry } from "./csrf"; +import { reportSearchFailure } from "./searchDiagnostics"; /** SSE event types emitted by streaming endpoints. */ const SSE_EVENT_STATUS = "status"; @@ -43,6 +45,9 @@ export class StreamFailureError extends Error { readonly code?: string; readonly retryable?: boolean; readonly stage?: string; + httpStatus?: number; + requestId?: string; + retryAfterSeconds?: number; constructor(streamError: StreamError, options?: ErrorOptions) { super(streamError.message, options); @@ -255,6 +260,7 @@ export async function streamSse( requestOptions: StreamSseRequestOptions = {}, ): Promise { const abortSignal = requestOptions.signal; + const startedAt = performance.now(); let httpResponse: Response; try { @@ -275,10 +281,16 @@ export async function streamSse( if (abortSignal?.aborted || isAbortError(fetchError)) { return; } + reportStreamFailure(url, startedAt, fetchError); throwNetworkStreamFailure(callbacks, source, fetchError); } - await consumeSseStream(httpResponse, callbacks, source, abortSignal); + try { + await consumeSseStream(httpResponse, callbacks, source, abortSignal); + } catch (streamFailure) { + reportStreamFailure(url, startedAt, streamFailure, httpResponse); + throw streamFailure; + } } /** @@ -291,6 +303,7 @@ export async function streamSseGet( requestOptions: StreamSseRequestOptions = {}, ): Promise { const abortSignal = requestOptions.signal; + const startedAt = performance.now(); let httpResponse: Response; try { @@ -299,10 +312,49 @@ export async function streamSseGet( if (abortSignal?.aborted || isAbortError(fetchError)) { return; } + reportStreamFailure(url, startedAt, fetchError); throwNetworkStreamFailure(callbacks, source, fetchError); } - await consumeSseStream(httpResponse, callbacks, source, abortSignal); + try { + await consumeSseStream(httpResponse, callbacks, source, abortSignal); + } catch (streamFailure) { + reportStreamFailure(url, startedAt, streamFailure, httpResponse); + throw streamFailure; + } +} + +function reportStreamFailure( + url: string, + startedAt: number, + failure: unknown, + httpResponse?: Response, +): void { + const streamFailure = failure instanceof StreamFailureError ? failure : undefined; + const frontendScript = document.querySelector('script[type="module"][src]'); + let failureKind: BrowserSearchFailure["kind"] = "PROTOCOL"; + if (!httpResponse) { + failureKind = "NETWORK"; + } else if (!httpResponse.ok) { + failureKind = "HTTP"; + } else if (streamFailure?.cause) { + failureKind = "NETWORK"; + } else if (streamFailure) { + failureKind = "SSE"; + } + void reportSearchFailure({ + path: new URL(url, window.location.origin).pathname, + kind: failureKind, + httpStatus: httpResponse?.status, + requestId: httpResponse?.headers.get("X-Request-ID") ?? undefined, + code: streamFailure?.code, + stage: streamFailure?.stage, + retryable: streamFailure?.retryable, + retryAfterSeconds: streamFailure?.retryAfterSeconds, + pagePath: window.location.pathname, + frontendAsset: frontendScript ? new URL(frontendScript.src).pathname : undefined, + elapsedMs: Math.round(performance.now() - startedAt), + }); } async function consumeSseStream( @@ -312,17 +364,31 @@ async function consumeSseStream( abortSignal?: AbortSignal, ): Promise { if (!httpResponse.ok) { - const apiMessage = await extractApiErrorMessage(httpResponse, `streamSse:${source}`); + const apiError = await readApiError(httpResponse, `streamSse:${source}`); + const apiMessage = apiError?.message.trim(); const statusSummary = `HTTP ${httpResponse.status}: ${httpResponse.statusText || "Request failed"}`; const httpFailure: StreamError = { - message: apiMessage ?? statusSummary, + message: apiMessage || statusSummary, retryable: isRetryableHttpStatus(httpResponse.status), }; if (apiMessage) { - httpFailure.details = statusSummary; + httpFailure.details = apiError?.details || statusSummary; } callbacks.onError?.(httpFailure); - throw new StreamFailureError(httpFailure); + const refusal = new StreamFailureError(httpFailure); + refusal.httpStatus = httpResponse.status; + refusal.requestId = httpResponse.headers.get("X-Request-ID") ?? undefined; + const retryAfter = httpResponse.headers.get("Retry-After"); + if (retryAfter) { + const retrySeconds = Number(retryAfter); + const retryDate = Date.parse(retryAfter); + if (Number.isFinite(retrySeconds) && retrySeconds >= 0) { + refusal.retryAfterSeconds = retrySeconds; + } else if (Number.isFinite(retryDate)) { + refusal.retryAfterSeconds = Math.max(0, Math.ceil((retryDate - Date.now()) / 1000)); + } + } + throw refusal; } const sseReader = httpResponse.body?.getReader(); diff --git a/frontend/src/lib/validation/schemas.ts b/frontend/src/lib/validation/schemas.ts index 6654f95c..4afd1c54 100644 --- a/frontend/src/lib/validation/schemas.ts +++ b/frontend/src/lib/validation/schemas.ts @@ -64,6 +64,23 @@ export const StreamStatusSchema = z.union([ /** Error response from SSE error events. */ export const StreamErrorSchema = z.object(sseEventFieldShape); +/** Bounded, credential-free observations from the browser's search transport. */ +export const BrowserSearchFailureSchema = z.object({ + path: z.string().startsWith("/api/").max(160), + kind: z.enum(["HTTP", "SSE", "NETWORK", "PROTOCOL"]), + httpStatus: z.number().int().min(100).max(599).optional(), + requestId: z.string().max(128).optional(), + code: z.string().max(128).optional(), + stage: z.string().max(64).optional(), + retryable: z.boolean().optional(), + retryAfterSeconds: z.number().nonnegative().optional(), + pagePath: z.string().max(160), + frontendAsset: z.string().max(160).optional(), + elapsedMs: z.number().nonnegative(), +}); + +export type BrowserSearchFailure = z.infer; + /** Canonical text chunk emitted by SSE streams. */ export const TextChunkSchema = z.object({ text: z.string(), diff --git a/src/main/java/com/williamcallahan/javachat/adapters/in/web/ApiRequestLoggingFilter.java b/src/main/java/com/williamcallahan/javachat/adapters/in/web/ApiRequestLoggingFilter.java new file mode 100644 index 00000000..676e4299 --- /dev/null +++ b/src/main/java/com/williamcallahan/javachat/adapters/in/web/ApiRequestLoggingFilter.java @@ -0,0 +1,80 @@ +package com.williamcallahan.javachat.adapters.in.web; + +import com.williamcallahan.javachat.support.StructuredLogValue; +import jakarta.servlet.FilterChain; +import jakarta.servlet.ServletException; +import jakarta.servlet.http.HttpServletRequest; +import jakarta.servlet.http.HttpServletResponse; +import java.io.IOException; +import java.util.concurrent.TimeUnit; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.slf4j.event.Level; +import org.springframework.core.Ordered; +import org.springframework.core.annotation.Order; +import org.springframework.http.HttpStatus; +import org.springframework.stereotype.Component; +import org.springframework.web.filter.OncePerRequestFilter; + +/** Correlates API requests, including security rejections and asynchronous streaming completion. */ +@Component +@Order(Ordered.HIGHEST_PRECEDENCE) +public final class ApiRequestLoggingFilter extends OncePerRequestFilter { + private static final Logger log = LoggerFactory.getLogger(ApiRequestLoggingFilter.class); + private static final String REQUEST_STARTED_ATTRIBUTE = ApiRequestLoggingFilter.class.getName() + ".started"; + private static final String REQUEST_ID_HEADER = "X-Request-ID"; + private static final int MAX_REQUEST_LOG_FIELD_LENGTH = 256; + + /** Excludes static assets and actuator scrapes from application request records. */ + @Override + protected boolean shouldNotFilter(HttpServletRequest apiRequest) { + return !apiRequest.getRequestURI().startsWith("/api/"); + } + + /** Includes the completion dispatch rather than timing only servlet setup. */ + @Override + protected boolean shouldNotFilterAsyncDispatch() { + return false; + } + + /** Issues a browser-visible ID before security can reject a request. */ + @Override + protected void doFilterInternal( + HttpServletRequest apiRequest, HttpServletResponse apiResponse, FilterChain apiFilterChain) + throws ServletException, IOException { + String requestId = apiRequest.getRequestId(); + String requestPath = StructuredLogValue.bounded(apiRequest.getRequestURI(), MAX_REQUEST_LOG_FIELD_LENGTH) + .text(); + if (apiRequest.getAttribute(REQUEST_STARTED_ATTRIBUTE) == null) { + apiRequest.setAttribute(REQUEST_STARTED_ATTRIBUTE, System.nanoTime()); + apiResponse.setHeader(REQUEST_ID_HEADER, requestId); + log.info( + "event=api_request_started requestId={} method={} path={}", + requestId, + apiRequest.getMethod(), + requestPath); + } + try { + apiFilterChain.doFilter(apiRequest, apiResponse); + } finally { + // Spring MVC performs a final ASYNC dispatch after an SSE emitter completes or fails. + if (!apiRequest.isAsyncStarted()) { + long startedNanos = (Long) apiRequest.getAttribute(REQUEST_STARTED_ATTRIBUTE); + log.atLevel(apiResponse.getStatus() >= HttpStatus.BAD_REQUEST.value() ? Level.WARN : Level.INFO) + .setMessage("API request completed") + .addKeyValue("event", "api_request_completed") + .addKeyValue("requestId", requestId) + .addKeyValue("method", apiRequest.getMethod()) + .addKeyValue("path", requestPath) + .addKeyValue("status", apiResponse.getStatus()) + .addKeyValue("durationMs", TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startedNanos)) + .addKeyValue( + "retryAfter", + StructuredLogValue.bounded( + apiResponse.getHeader("Retry-After"), MAX_REQUEST_LOG_FIELD_LENGTH) + .text()) + .log(); + } + } + } +} diff --git a/src/main/java/com/williamcallahan/javachat/adapters/in/web/BrowserSearchFailureController.java b/src/main/java/com/williamcallahan/javachat/adapters/in/web/BrowserSearchFailureController.java new file mode 100644 index 00000000..536e27b0 --- /dev/null +++ b/src/main/java/com/williamcallahan/javachat/adapters/in/web/BrowserSearchFailureController.java @@ -0,0 +1,84 @@ +package com.williamcallahan.javachat.adapters.in.web; + +import com.williamcallahan.javachat.support.StructuredLogValue; +import jakarta.validation.Valid; +import jakarta.validation.constraints.Max; +import jakarta.validation.constraints.Min; +import jakarta.validation.constraints.NotBlank; +import jakarta.validation.constraints.NotNull; +import jakarta.validation.constraints.Pattern; +import jakarta.validation.constraints.PositiveOrZero; +import jakarta.validation.constraints.Size; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.http.ResponseEntity; +import org.springframework.web.bind.annotation.PostMapping; +import org.springframework.web.bind.annotation.RequestBody; +import org.springframework.web.bind.annotation.RestController; + +/** Retains browser-observed search failures that cannot be reconstructed from origin HTTP statuses. */ +@RestController +public final class BrowserSearchFailureController { + private static final Logger log = LoggerFactory.getLogger(BrowserSearchFailureController.class); + private static final int MAX_DIAGNOSTIC_FIELD_LENGTH = 160; + + /** Records a CSRF-protected, explicitly untrusted observation without accepting prompts or credentials. */ + @PostMapping("/api/diagnostics/search-failures") + public ResponseEntity recordFailure(@Valid @RequestBody SearchFailureReport failureReport) { + log.atWarn() + .setMessage("Browser reported a search failure") + .addKeyValue("event", "browser_search_failed") + .addKeyValue("provenance", "client_reported") + .addKeyValue("path", failureReport.path()) + .addKeyValue("kind", failureReport.kind()) + .addKeyValue("httpStatus", failureReport.httpStatus()) + .addKeyValue( + "requestId", + StructuredLogValue.bounded(failureReport.requestId(), MAX_DIAGNOSTIC_FIELD_LENGTH) + .text()) + .addKeyValue( + "code", + StructuredLogValue.bounded(failureReport.code(), MAX_DIAGNOSTIC_FIELD_LENGTH) + .text()) + .addKeyValue( + "stage", + StructuredLogValue.bounded(failureReport.stage(), MAX_DIAGNOSTIC_FIELD_LENGTH) + .text()) + .addKeyValue("retryable", failureReport.retryable()) + .addKeyValue("retryAfterSeconds", failureReport.retryAfterSeconds()) + .addKeyValue("pagePath", failureReport.pagePath()) + .addKeyValue("frontendAsset", failureReport.frontendAsset()) + .addKeyValue("elapsedMs", failureReport.elapsedMs()) + .log(); + return ResponseEntity.accepted().build(); + } + + /** Distinguishes a received refusal from an SSE failure or a browser transport/protocol failure. */ + public enum FailureKind { + HTTP, + SSE, + NETWORK, + PROTOCOL + } + + /** Defines only the bounded metadata the browser is allowed to send into application logs. */ + public record SearchFailureReport( + @NotBlank @Size(max = 160) @Pattern(regexp = "^/api/[^?#\\r\\n]*$") + String path, + + @NotNull FailureKind kind, + @Min(100) @Max(599) Integer httpStatus, + @Size(max = 128) String requestId, + @Size(max = 128) String code, + @Size(max = 64) String stage, + Boolean retryable, + @PositiveOrZero Double retryAfterSeconds, + + @NotBlank @Size(max = 160) @Pattern(regexp = "^/[^?#\\r\\n]*$") + String pagePath, + + @Size(max = 160) @Pattern(regexp = "^/[^?#\\r\\n]*$") + String frontendAsset, + + @PositiveOrZero long elapsedMs) {} +} diff --git a/src/main/java/com/williamcallahan/javachat/adapters/out/llm/openai/OpenAiStreamingFailureException.java b/src/main/java/com/williamcallahan/javachat/adapters/out/llm/openai/OpenAiStreamingFailureException.java index 8a019a37..f2e8049e 100644 --- a/src/main/java/com/williamcallahan/javachat/adapters/out/llm/openai/OpenAiStreamingFailureException.java +++ b/src/main/java/com/williamcallahan/javachat/adapters/out/llm/openai/OpenAiStreamingFailureException.java @@ -19,7 +19,8 @@ * Captures bounded terminal context from an OpenAI-compatible streaming provider. * *

The exception retains the upstream cause for recovery decisions but emits exactly one structured alert - * without projecting upstream bodies, arbitrary headers, or request identifiers.

+ * without projecting upstream bodies or arbitrary headers. Gateway correlation and retry headers + * remain available for joining the provider refusal to its retained request.

*/ public final class OpenAiStreamingFailureException extends RuntimeException implements ReportedStreamingFailure { @Serial @@ -43,6 +44,8 @@ enum AlertField { UPSTREAM_HTTP_STATUS("upstreamHttpStatus"), UPSTREAM_CODE("upstreamCode"), UPSTREAM_TYPE("upstreamType"), + GATEWAY_REQUEST_ID("gatewayRequestId"), + RETRY_AFTER("retryAfter"), QUEUE_TIER("queueTier"), QUEUE_LANE("queueLane"), QUEUE_TIMEOUT_PHASE("queueTimeoutPhase"), @@ -112,6 +115,8 @@ String wireName() { private final Integer upstreamHttpStatus; private final String upstreamCode; private final String upstreamType; + private final String gatewayRequestId; + private final String retryAfter; private final Map queueHeaderValues; private OpenAiStreamingFailureException(Throwable upstreamFailure, TerminalAttempt terminalAttempt) { @@ -128,11 +133,19 @@ private OpenAiStreamingFailureException(Throwable upstreamFailure, TerminalAttem this.upstreamHttpStatus = serviceException.statusCode(); this.upstreamCode = boundedOptionalValue(serviceException.code()).orElse(null); this.upstreamType = boundedOptionalValue(serviceException.type()).orElse(null); + this.gatewayRequestId = boundedOptionalValue(serviceException.headers().values("x-lgw-request-id").stream() + .findFirst()) + .orElse(null); + this.retryAfter = boundedOptionalValue(serviceException.headers().values("retry-after").stream() + .findFirst()) + .orElse(null); this.queueHeaderValues = captureQueueHeaderValues(serviceException); } else { this.upstreamHttpStatus = null; this.upstreamCode = null; this.upstreamType = null; + this.gatewayRequestId = null; + this.retryAfter = null; this.queueHeaderValues = Map.of(); } this.failureKind = determineFailureKind(upstreamFailure, Optional.ofNullable(upstreamCode)); @@ -251,6 +264,12 @@ private void logTerminalAlert() { if (upstreamType != null) { terminalAlert = terminalAlert.addKeyValue(AlertField.UPSTREAM_TYPE.fieldName(), upstreamType); } + if (gatewayRequestId != null) { + terminalAlert = terminalAlert.addKeyValue(AlertField.GATEWAY_REQUEST_ID.fieldName(), gatewayRequestId); + } + if (retryAfter != null) { + terminalAlert = terminalAlert.addKeyValue(AlertField.RETRY_AFTER.fieldName(), retryAfter); + } for (QueueHeader queueHeader : QueueHeader.values()) { if (queueHeaderValues.containsKey(queueHeader)) { terminalAlert = terminalAlert.addKeyValue( diff --git a/src/main/java/com/williamcallahan/javachat/service/OpenAiCompatibleEmbeddingClient.java b/src/main/java/com/williamcallahan/javachat/service/OpenAiCompatibleEmbeddingClient.java index 082ef238..e243e903 100644 --- a/src/main/java/com/williamcallahan/javachat/service/OpenAiCompatibleEmbeddingClient.java +++ b/src/main/java/com/williamcallahan/javachat/service/OpenAiCompatibleEmbeddingClient.java @@ -9,6 +9,7 @@ import com.openai.errors.OpenAIServiceException; import com.openai.models.embeddings.CreateEmbeddingResponse; import com.openai.models.embeddings.EmbeddingCreateParams; +import com.williamcallahan.javachat.support.StructuredLogValue; import java.io.InterruptedIOException; import java.net.SocketTimeoutException; import java.time.Duration; @@ -357,7 +358,7 @@ private List createEmbeddings( .build())); CompletableFuture accountedEmbeddingResponseFuture = embeddingResponseFuture.whenComplete((completedEmbeddingResponse, embeddingFailure) -> { - recordProviderCooldownFromFutureFailure(embeddingFailure, requestTier); + recordProviderCooldownFromFutureFailure(embeddingFailure, requestTier, purpose); requestPermits.release(); }); requestPermitAcquired = false; @@ -367,7 +368,7 @@ private List createEmbeddings( throw new EmbeddingServiceTemporarilyUnavailableException( "Interrupted during gateway embedding request admission or completion", interruptedException); } catch (OpenAIServiceException exception) { - recordProviderCooldown(exception, requestTier); + recordProviderCooldown(exception, requestTier, purpose); throw wrapServiceError(exception); } catch (OpenAIRetryableException exception) { throw wrapRetryableError(exception); @@ -432,10 +433,9 @@ private EmbeddingServiceUnavailableException wrapServiceError(OpenAIServiceExcep exception, Duration.ofSeconds(retryDecision.retryAfterSeconds())); } catch (RateLimitDecisionException rateLimitDecisionFailure) { - rateLimitDecisionFailure.addSuppressed(exception); + exception.addSuppressed(rateLimitDecisionFailure); return new EmbeddingServiceTemporarilyUnavailableException( - "Remote embedding provider returned HTTP 429 without authoritative retry timing", - rateLimitDecisionFailure); + "Remote embedding provider returned HTTP 429 without authoritative retry timing", exception); } } String details = sanitizeMessage(exception.getMessage()); @@ -451,20 +451,54 @@ private EmbeddingServiceUnavailableException wrapServiceError(OpenAIServiceExcep return new EmbeddingServiceUnavailableException(failureMessage, exception); } - private void recordProviderCooldownFromFutureFailure(Throwable embeddingFailure, LlmGatewayTier requestTier) { + private void recordProviderCooldownFromFutureFailure( + Throwable embeddingFailure, LlmGatewayTier requestTier, LlmGatewayTier.Purpose purpose) { Throwable providerFailure = embeddingFailure; while (providerFailure instanceof CompletionException && providerFailure.getCause() != null) { providerFailure = providerFailure.getCause(); } if (providerFailure instanceof OpenAIServiceException serviceException) { - recordProviderCooldown(serviceException, requestTier); + recordProviderCooldown(serviceException, requestTier, purpose); } } - private void recordProviderCooldown(OpenAIServiceException serviceException, LlmGatewayTier requestTier) { + private void recordProviderCooldown( + OpenAIServiceException serviceException, LlmGatewayTier requestTier, LlmGatewayTier.Purpose purpose) { if (serviceException.statusCode() != HttpStatus.TOO_MANY_REQUESTS.value()) { return; } + log.atWarn() + .setMessage("Embedding provider rejected request") + .addKeyValue("event", "embedding_rate_limited") + .addKeyValue("upstreamHttpStatus", serviceException.statusCode()) + .addKeyValue("tier", requestTier.requestHeader()) + .addKeyValue("model", modelName) + .addKeyValue("workload", requestTier.workloadHeader(purpose)) + .addKeyValue( + "upstreamCode", + StructuredLogValue.bounded(serviceException.code().orElse("unknown"), MAX_ERROR_SNIPPET) + .text()) + .addKeyValue( + "upstreamType", + StructuredLogValue.bounded(serviceException.type().orElse("unknown"), MAX_ERROR_SNIPPET) + .text()) + .addKeyValue( + "retryAfter", + StructuredLogValue.bounded( + serviceException.headers().values("retry-after").stream() + .findFirst() + .orElse("unknown"), + MAX_ERROR_SNIPPET) + .text()) + .addKeyValue( + "gatewayRequestId", + StructuredLogValue.bounded( + serviceException.headers().values("x-lgw-request-id").stream() + .findFirst() + .orElse("unknown"), + MAX_ERROR_SNIPPET) + .text()) + .log(); try { RateLimitDecision rateLimitDecision = rateLimitDecisionResolver.resolveFromOpenAiRetryAfterHeaders(serviceException.headers()); diff --git a/src/main/java/com/williamcallahan/javachat/web/ChatController.java b/src/main/java/com/williamcallahan/javachat/web/ChatController.java index 8b0b810f..3efb669d 100644 --- a/src/main/java/com/williamcallahan/javachat/web/ChatController.java +++ b/src/main/java/com/williamcallahan/javachat/web/ChatController.java @@ -228,55 +228,62 @@ public Flux> stream( Flux> operationEventsWithProgress = Flux.merge( retrievalProgressEvents.asFlux(), deadlineBoundOperationEvents.doFinally(terminationSignal -> retrievalProgressEvents.tryEmitComplete())); - return Flux.concat( - sseSupport.responsePreparationStatus(), sseSupport.withHeartbeats(operationEventsWithProgress)) - .onErrorResume(error -> { - Optional terminalFailureContext = - ReportedStreamingFailure.findInCauseChain(error); - Throwable upstreamError = terminalFailureContext - .map(ReportedStreamingFailure::upstreamFailure) - .orElse(error); - if (upstreamError instanceof ConfiguredProviderTemporarilyUnavailableException) { - return sseSupport.configuredProviderUnavailableError(); - } - if (terminalFailureContext.isEmpty() && sseSupport.isResponsePreparationTimeout(upstreamError)) { - PIPELINE_LOG - .atWarn() - .setMessage("Response preparation timeout") - .addKeyValue("requestToken", requestToken) - .addKeyValue( - "sessionId", - StructuredLogValue.bounded(sessionId, MAX_STREAM_LOG_SESSION_ID_LENGTH) - .text()) - .addKeyValue("code", STATUS_CODE_RETRIEVAL_TIMEOUT) - .addKeyValue("stage", STATUS_STAGE_RETRIEVAL) - .addKeyValue( - "exceptionType", - upstreamError.getClass().getSimpleName()) - .log(); - return sseSupport.responsePreparationTimeoutError(); - } - String errorDetail = buildUserFacingErrorMessage(upstreamError); - String diagnostics = isRetrievalFailure(upstreamError) - ? RETRIEVAL_UNAVAILABLE_DETAILS - : upstreamError instanceof Exception exception - ? describeException(exception) - : upstreamError.getClass().getName(); - if (terminalFailureContext.isEmpty()) { - PIPELINE_LOG - .atError() - .setMessage("[{}] STREAMING ERROR") - .addArgument(requestToken) - .addKeyValue( - "sessionId", - StructuredLogValue.bounded(sessionId, MAX_STREAM_LOG_SESSION_ID_LENGTH) - .text()) - .addKeyValue("exceptionType", error.getClass().getSimpleName()) - .log(); - } - boolean retryable = openAIStreamingService.isRecoverableStreamingFailure(error); - return sseSupport.streamErrorEvent(errorDetail, diagnostics, retryable); - }); + return sseSupport.withRequestDiagnostics( + Flux.concat( + sseSupport.responsePreparationStatus(), + sseSupport.withHeartbeats(operationEventsWithProgress)) + .onErrorResume(error -> { + Optional terminalFailureContext = + ReportedStreamingFailure.findInCauseChain(error); + Throwable upstreamError = terminalFailureContext + .map(ReportedStreamingFailure::upstreamFailure) + .orElse(error); + if (upstreamError instanceof ConfiguredProviderTemporarilyUnavailableException) { + return sseSupport.configuredProviderUnavailableError(); + } + if (terminalFailureContext.isEmpty() + && sseSupport.isResponsePreparationTimeout(upstreamError)) { + PIPELINE_LOG + .atWarn() + .setMessage("Response preparation timeout") + .addKeyValue("requestToken", requestToken) + .addKeyValue( + "sessionId", + StructuredLogValue.bounded(sessionId, MAX_STREAM_LOG_SESSION_ID_LENGTH) + .text()) + .addKeyValue("code", STATUS_CODE_RETRIEVAL_TIMEOUT) + .addKeyValue("stage", STATUS_STAGE_RETRIEVAL) + .addKeyValue( + "exceptionType", + upstreamError.getClass().getSimpleName()) + .log(); + return sseSupport.responsePreparationTimeoutError(); + } + String errorDetail = buildUserFacingErrorMessage(upstreamError); + String diagnostics = isRetrievalFailure(upstreamError) + ? RETRIEVAL_UNAVAILABLE_DETAILS + : upstreamError instanceof Exception exception + ? describeException(exception) + : upstreamError.getClass().getName(); + if (terminalFailureContext.isEmpty()) { + PIPELINE_LOG + .atError() + .setMessage("[{}] STREAMING ERROR") + .addArgument(requestToken) + .addKeyValue( + "sessionId", + StructuredLogValue.bounded(sessionId, MAX_STREAM_LOG_SESSION_ID_LENGTH) + .text()) + .addKeyValue( + "exceptionType", + error.getClass().getSimpleName()) + .log(); + } + boolean retryable = openAIStreamingService.isRecoverableStreamingFailure(error); + return sseSupport.streamErrorEvent(errorDetail, diagnostics, retryable); + }), + response, + "/api/chat/stream"); } /** diff --git a/src/main/java/com/williamcallahan/javachat/web/GuidedLearningController.java b/src/main/java/com/williamcallahan/javachat/web/GuidedLearningController.java index 4b9b174b..2850e7e4 100644 --- a/src/main/java/com/williamcallahan/javachat/web/GuidedLearningController.java +++ b/src/main/java/com/williamcallahan/javachat/web/GuidedLearningController.java @@ -152,27 +152,33 @@ public Flux> streamLesson( @RequestParam(name = "slug", required = false) String slug, HttpServletResponse response) { Flux lessonContentStream = curatedLessonContentStream(slug); sseSupport.configureStreamingHeaders(response); - return Flux.defer(() -> { - Flux dataStream = sseSupport.prepareDataStream(lessonContentStream, ignoredChunk -> {}); - return sseSupport.withHeartbeats(dataStream.map(sseSupport::textEvent)); - }) - .onErrorResume(error -> { - if (ReportedStreamingFailure.findInCauseChain(error).isEmpty()) { - log.atError() - .setMessage("Guided lesson content stream error") - .addKeyValue( - "lessonSlug", - StructuredLogValue.bounded(slug, MAX_GUIDED_LOG_FIELD_LENGTH) - .text()) - .addKeyValue("exceptionType", error.getClass().getSimpleName()) - .log(); - } - boolean retryable = openAIStreamingService.isRecoverableStreamingFailure(error); - return sseSupport.streamErrorEvent( - "Lesson content stream failed", - "The lesson content stream encountered an error. Please try again.", - retryable); - }); + return sseSupport.withRequestDiagnostics( + Flux.defer(() -> { + Flux dataStream = + sseSupport.prepareDataStream(lessonContentStream, ignoredChunk -> {}); + return sseSupport.withHeartbeats(dataStream.map(sseSupport::textEvent)); + }) + .onErrorResume(error -> { + if (ReportedStreamingFailure.findInCauseChain(error).isEmpty()) { + log.atError() + .setMessage("Guided lesson content stream error") + .addKeyValue( + "lessonSlug", + StructuredLogValue.bounded(slug, MAX_GUIDED_LOG_FIELD_LENGTH) + .text()) + .addKeyValue( + "exceptionType", + error.getClass().getSimpleName()) + .log(); + } + boolean retryable = openAIStreamingService.isRecoverableStreamingFailure(error); + return sseSupport.streamErrorEvent( + "Lesson content stream failed", + "The lesson content stream encountered an error. Please try again.", + retryable); + }), + response, + "/api/guided/content/stream"); } /** @@ -297,9 +303,13 @@ public Flux> stream( Flux> operationEventsWithProgress = Flux.merge( retrievalProgressEvents.asFlux(), deadlineBoundOperationEvents.doFinally(terminationSignal -> retrievalProgressEvents.tryEmitComplete())); - return Flux.concat( - sseSupport.responsePreparationStatus(), sseSupport.withHeartbeats(operationEventsWithProgress)) - .onErrorResume(error -> guidedStreamError(error, requestToken, sessionId, canonicalLessonSlug)); + return sseSupport.withRequestDiagnostics( + Flux.concat( + sseSupport.responsePreparationStatus(), + sseSupport.withHeartbeats(operationEventsWithProgress)) + .onErrorResume(error -> guidedStreamError(error, requestToken, sessionId, canonicalLessonSlug)), + response, + "/api/guided/stream"); } /** diff --git a/src/main/java/com/williamcallahan/javachat/web/SseSupport.java b/src/main/java/com/williamcallahan/javachat/web/SseSupport.java index c16921ec..88f87743 100644 --- a/src/main/java/com/williamcallahan/javachat/web/SseSupport.java +++ b/src/main/java/com/williamcallahan/javachat/web/SseSupport.java @@ -125,6 +125,17 @@ public void configureStreamingHeaders(HttpServletResponse response) { response.addHeader(HttpHeaders.CACHE_CONTROL, "no-cache, no-transform"); } + /** Carries request correlation across the asynchronous preparation and provider callbacks. */ + public Flux> withRequestDiagnostics( + Flux> streamEvents, HttpServletResponse servletResponse, String requestPath) { + StreamRequestObservation observation = + new StreamRequestObservation(servletResponse.getHeader("X-Request-ID"), requestPath, System.nanoTime()); + return streamEvents.contextWrite( + streamContext -> streamContext.put(StreamRequestObservation.class, observation)); + } + + private record StreamRequestObservation(String requestId, String path, long startedNanos) {} + /** * Prepares a raw content stream for SSE transmission with backpressure handling. * Filters empty chunks and applies the standard streaming configuration. @@ -232,8 +243,26 @@ public Flux> sseError(SseEventPayload payload) { log.error("Failed to serialize SSE error payload", serializationFailure); json = ERROR_FALLBACK_JSON; } - return Flux.just( - ServerSentEvent.builder().event(EVENT_ERROR).data(json).build()); + ServerSentEvent errorEvent = + ServerSentEvent.builder().event(EVENT_ERROR).data(json).build(); + return Flux.deferContextual(streamContext -> { + streamContext + .getOrEmpty(StreamRequestObservation.class) + .ifPresent(observation -> log.atWarn() + .setMessage("Search stream emitted a terminal error") + .addKeyValue("event", "search_stream_failed") + .addKeyValue("requestId", observation.requestId()) + .addKeyValue("path", observation.path()) + .addKeyValue("code", payload.code()) + .addKeyValue("stage", payload.stage()) + .addKeyValue("retryable", payload.retryable()) + .addKeyValue( + "durationMs", + java.util.concurrent.TimeUnit.NANOSECONDS.toMillis( + System.nanoTime() - observation.startedNanos())) + .log()); + return Flux.just(errorEvent); + }); } /** diff --git a/src/test/java/com/williamcallahan/javachat/adapters/in/web/BrowserSearchFailureControllerTest.java b/src/test/java/com/williamcallahan/javachat/adapters/in/web/BrowserSearchFailureControllerTest.java new file mode 100644 index 00000000..99d8d3d9 --- /dev/null +++ b/src/test/java/com/williamcallahan/javachat/adapters/in/web/BrowserSearchFailureControllerTest.java @@ -0,0 +1,80 @@ +package com.williamcallahan.javachat.adapters.in.web; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.springframework.security.test.web.servlet.request.SecurityMockMvcRequestPostProcessors.csrf; +import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post; +import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; + +import ch.qos.logback.classic.Logger; +import com.williamcallahan.javachat.adapters.out.clerk.ClerkApiKeyVerifier; +import com.williamcallahan.javachat.config.AppProperties; +import com.williamcallahan.javachat.config.SecurityConfig; +import com.williamcallahan.javachat.support.logging.ExpectedLogEvents; +import org.junit.jupiter.api.Test; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.autoconfigure.web.servlet.WebMvcTest; +import org.springframework.context.annotation.Import; +import org.springframework.http.MediaType; +import org.springframework.test.context.bean.override.mockito.MockitoBean; +import org.springframework.test.web.servlet.MockMvc; + +/** Proves browser observations enter retained logs only through the bounded CSRF-protected boundary. */ +@WebMvcTest(BrowserSearchFailureController.class) +@Import({AppProperties.class, SecurityConfig.class}) +class BrowserSearchFailureControllerTest { + private static final String DIAGNOSTICS_ENDPOINT = "/api/diagnostics/search-failures"; + private static final String REFUSAL_REPORT = """ + {"path":"/api/chat/stream","kind":"HTTP","httpStatus":429, + "requestId":"search-request-test","code":"diagnostic.probe","retryAfterSeconds":12, + "pagePath":"/chat","frontendAsset":"/assets/app-test.js","elapsedMs":10} + """; + + @Autowired + MockMvc mockMvc; + + @MockitoBean + ClerkApiKeyVerifier clerkApiKeyVerifier; + + @Test + void retainsRefusalMetadataAsClientReportedWithoutPromotingItToVerifiedServerFailure() throws Exception { + try (ExpectedLogEvents capturedLogs = + ExpectedLogEvents.capture((Logger) LoggerFactory.getLogger(BrowserSearchFailureController.class))) { + mockMvc.perform(post(DIAGNOSTICS_ENDPOINT) + .with(csrf()) + .contentType(MediaType.APPLICATION_JSON) + .content(REFUSAL_REPORT)) + .andExpect(status().isAccepted()); + assertEquals(1, capturedLogs.events().size()); + assertTrue(capturedLogs.events().getFirst().getKeyValuePairs().stream() + .anyMatch( + logField -> "provenance".equals(logField.key) && "client_reported".equals(logField.value))); + assertTrue(capturedLogs.events().getFirst().getKeyValuePairs().stream() + .anyMatch(logField -> "httpStatus".equals(logField.key) + && Integer.valueOf(429).equals(logField.value))); + } + } + + @Test + void requiresCsrfProtection() throws Exception { + mockMvc.perform(post(DIAGNOSTICS_ENDPOINT) + .contentType(MediaType.APPLICATION_JSON) + .content(REFUSAL_REPORT)) + .andExpect(status().isForbidden()); + } + + @Test + void rejectsQueryStringsAndOversizedDiagnosticFields() throws Exception { + mockMvc.perform(post(DIAGNOSTICS_ENDPOINT) + .with(csrf()) + .contentType(MediaType.APPLICATION_JSON) + .content(REFUSAL_REPORT.replace("/api/chat/stream", "/api/chat/stream?secret=excluded"))) + .andExpect(status().isBadRequest()); + mockMvc.perform(post(DIAGNOSTICS_ENDPOINT) + .with(csrf()) + .contentType(MediaType.APPLICATION_JSON) + .content(REFUSAL_REPORT.replace("diagnostic.probe", "x".repeat(129)))) + .andExpect(status().isBadRequest()); + } +} diff --git a/src/test/java/com/williamcallahan/javachat/adapters/out/llm/openai/OpenAiStreamingFailureExceptionTest.java b/src/test/java/com/williamcallahan/javachat/adapters/out/llm/openai/OpenAiStreamingFailureExceptionTest.java index e42615a3..6d1ccf4b 100644 --- a/src/test/java/com/williamcallahan/javachat/adapters/out/llm/openai/OpenAiStreamingFailureExceptionTest.java +++ b/src/test/java/com/williamcallahan/javachat/adapters/out/llm/openai/OpenAiStreamingFailureExceptionTest.java @@ -55,6 +55,8 @@ void terminalFailureCapturesGatewayTimeoutContextAfterSecondAttempt() { .put(QueueHeader.TIMEOUT_PHASE.headerName(), "first_output") .put(QueueHeader.TIMEOUT_SECONDS.headerName(), "75") .put(QueueHeader.ERROR_SOURCE.headerName(), "queue") + .put("x-lgw-request-id", "gateway-correlation-test") + .put("Retry-After", "12") .build(); ErrorObject gatewayError = ErrorObject.builder() .message(UPSTREAM_SECRET_MESSAGE) @@ -110,6 +112,8 @@ void terminalFailureCapturesGatewayTimeoutContextAfterSecondAttempt() { assertLogField(terminalAlert, AlertField.UPSTREAM_HTTP_STATUS, UPSTREAM_GATEWAY_TIMEOUT_STATUS); assertLogField(terminalAlert, AlertField.UPSTREAM_CODE, StreamingFailureKind.QUEUE_UPSTREAM_TIMEOUT.wireName()); assertLogField(terminalAlert, AlertField.UPSTREAM_TYPE, "upstream_timeout"); + assertLogField(terminalAlert, AlertField.GATEWAY_REQUEST_ID, "gateway-correlation-test"); + assertLogField(terminalAlert, AlertField.RETRY_AFTER, "12"); assertLogField(terminalAlert, QueueHeader.TIER.alertField(), "production-z? event=?forged?next"); assertLogField(terminalAlert, QueueHeader.LANE.alertField(), "local"); assertLogField(terminalAlert, QueueHeader.TIMEOUT_PHASE.alertField(), "first_output"); diff --git a/src/test/java/com/williamcallahan/javachat/config/SecurityConfigTest.java b/src/test/java/com/williamcallahan/javachat/config/SecurityConfigTest.java index 569c3116..ab0511ea 100644 --- a/src/test/java/com/williamcallahan/javachat/config/SecurityConfigTest.java +++ b/src/test/java/com/williamcallahan/javachat/config/SecurityConfigTest.java @@ -6,21 +6,27 @@ import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.Mockito.when; +import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.asyncDispatch; import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.get; import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.head; import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post; import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.content; import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.header; import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.jsonPath; +import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.request; import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; +import ch.qos.logback.classic.Logger; +import com.williamcallahan.javachat.adapters.in.web.ApiRequestLoggingFilter; import com.williamcallahan.javachat.adapters.out.clerk.ClerkApiKeyVerifier; import com.williamcallahan.javachat.application.auth.ApiKeyOperationUnavailableException; import com.williamcallahan.javachat.application.auth.VerifiedApiKey; +import com.williamcallahan.javachat.support.logging.ExpectedLogEvents; import com.williamcallahan.javachat.web.CsrfController; import jakarta.servlet.http.Cookie; import java.util.Optional; import org.junit.jupiter.api.Test; +import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.autoconfigure.web.servlet.WebMvcTest; import org.springframework.context.annotation.Import; @@ -31,8 +37,10 @@ import org.springframework.test.context.bean.override.mockito.MockitoBean; import org.springframework.test.web.servlet.MockMvc; import org.springframework.test.web.servlet.MvcResult; +import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.PostMapping; import org.springframework.web.bind.annotation.RestController; +import reactor.core.publisher.Flux; /** * Verifies that the browser-readable CSRF cookie remains valid across stateless requests. @@ -134,6 +142,47 @@ void authenticatesApiKeyBearerWithoutCsrfExchange() throws Exception { .andExpect(content().string(API_KEY_SUBJECT)); } + @Test + void identifiesRateLimitedRequestsBeforeControllerDispatch() throws Exception { + when(clerkApiKeyVerifier.verify(API_KEY_SECRET)).thenReturn(Optional.empty()); + for (int requestNumber = 0; requestNumber < 20; requestNumber++) { + mockMvc.perform(post(CSRF_PROTECTED_ENDPOINT) + .header(HttpHeaders.AUTHORIZATION, API_KEY_BEARER) + .with(httpRequest -> { + httpRequest.setRemoteAddr("192.0.2.20"); + return httpRequest; + })) + .andExpect(status().isUnauthorized()); + } + mockMvc.perform(post(CSRF_PROTECTED_ENDPOINT) + .header(HttpHeaders.AUTHORIZATION, API_KEY_BEARER) + .with(httpRequest -> { + httpRequest.setRemoteAddr("192.0.2.20"); + return httpRequest; + })) + .andExpect(status().isTooManyRequests()) + .andExpect(header().exists("X-Request-ID")); + } + + @Test + void recordsStreamingCompletionAfterTheAsynchronousResponseFinishes() throws Exception { + try (ExpectedLogEvents capturedLogs = + ExpectedLogEvents.capture((Logger) LoggerFactory.getLogger(ApiRequestLoggingFilter.class))) { + MvcResult streamingExchange = mockMvc.perform(get("/api/test/stream")) + .andExpect(request().asyncStarted()) + .andExpect(header().exists("X-Request-ID")) + .andReturn(); + assertTrue(capturedLogs.events().stream() + .noneMatch(logEvent -> "API request completed".equals(logEvent.getFormattedMessage()))); + mockMvc.perform(asyncDispatch(streamingExchange)).andExpect(status().isOk()); + assertEquals( + 1, + capturedLogs.events().stream() + .filter(logEvent -> "API request completed".equals(logEvent.getFormattedMessage())) + .count()); + } + } + @Test void returnsJsonWhenCsrfTokenIsMissing() throws Exception { mockMvc.perform(post(CSRF_PROTECTED_ENDPOINT)) @@ -252,6 +301,12 @@ private static void assertBrowserReadableCsrfCookieAttributes(Cookie csrfCookie) @RestController public static final class ProtectedPostController { + /** Exercises the real asynchronous servlet completion rather than a mocked filter chain. */ + @GetMapping(value = "/api/test/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE) + public Flux streamForRequestLogging() { + return Flux.just("complete"); + } + /** * Confirms that Spring Security admitted a state-changing request. * diff --git a/src/test/java/com/williamcallahan/javachat/service/OpenAiCompatibleEmbeddingClientTest.java b/src/test/java/com/williamcallahan/javachat/service/OpenAiCompatibleEmbeddingClientTest.java index 187bc7d3..fbe02ecb 100644 --- a/src/test/java/com/williamcallahan/javachat/service/OpenAiCompatibleEmbeddingClientTest.java +++ b/src/test/java/com/williamcallahan/javachat/service/OpenAiCompatibleEmbeddingClientTest.java @@ -15,6 +15,7 @@ import static org.mockito.Mockito.verifyNoInteractions; import static org.mockito.Mockito.when; +import ch.qos.logback.classic.Logger; import com.openai.client.OpenAIClient; import com.openai.client.OpenAIClientAsync; import com.openai.core.RequestOptions; @@ -24,6 +25,7 @@ import com.openai.models.embeddings.CreateEmbeddingResponse; import com.openai.models.embeddings.EmbeddingCreateParams; import com.openai.services.async.EmbeddingServiceAsync; +import com.williamcallahan.javachat.support.logging.ExpectedLogEvents; import java.io.InterruptedIOException; import java.time.Duration; import java.util.List; @@ -40,6 +42,7 @@ import java.util.concurrent.atomic.AtomicLongArray; import org.junit.jupiter.api.Test; import org.mockito.ArgumentCaptor; +import org.slf4j.LoggerFactory; /** * Verifies OpenAI embedding responses preserve request ordering. @@ -868,6 +871,68 @@ void preservesAuthoritativeRetryTimingForProviderAndLocalCooldownFailures() { } } + @Test + void recordsProviderRateLimitWithTimingAndRequestIdentity() { + OpenAIClient liveClient = mock(OpenAIClient.class); + OpenAIClient batchClient = mock(OpenAIClient.class); + EmbeddingServiceAsync liveEmbeddingService = mockAsyncEmbeddingService(liveClient); + RateLimitException providerRejection = RateLimitException.builder() + .headers(Headers.builder() + .put("Retry-After", "60") + .put("x-lgw-request-id", "embedding-rejection-test") + .build()) + .build(); + when(liveEmbeddingService.create(any(), any(RequestOptions.class))) + .thenReturn(CompletableFuture.failedFuture(providerRejection)); + + try (ExpectedLogEvents capturedLogs = ExpectedLogEvents.capture( + (Logger) LoggerFactory.getLogger(OpenAiCompatibleEmbeddingClient.class)); + OpenAiCompatibleEmbeddingClient embeddingClient = + new OpenAiCompatibleEmbeddingClient(liveClient, batchClient, gatewaySettings())) { + EmbeddingServiceTemporarilyUnavailableException embeddingFailure = assertThrows( + EmbeddingServiceTemporarilyUnavailableException.class, + () -> embeddingClient.embed( + LlmGatewayTier.Purpose.SEARCH_QUERY_EMBEDDING, + List.of("test question"), + LlmGatewayTier.LIVE)); + assertSame(providerRejection, embeddingFailure.getCause()); + assertTrue(capturedLogs.events().stream() + .anyMatch(logEvent -> logEvent.getFormattedMessage().contains("Embedding provider rejected request") + && logEvent.getKeyValuePairs().stream() + .anyMatch(logField -> "upstreamHttpStatus".equals(logField.key) + && Integer.valueOf(429).equals(logField.value)) + && logEvent.getKeyValuePairs().stream() + .anyMatch(logField -> "gatewayRequestId".equals(logField.key) + && "embedding-rejection-test".equals(logField.value)))); + } + } + + @Test + void missingRetryTimingDoesNotHideTheUpstreamRateLimitCause() { + OpenAIClient liveClient = mock(OpenAIClient.class); + OpenAIClient batchClient = mock(OpenAIClient.class); + EmbeddingServiceAsync liveEmbeddingService = mockAsyncEmbeddingService(liveClient); + RateLimitException providerRejection = + RateLimitException.builder().headers(Headers.builder().build()).build(); + when(liveEmbeddingService.create(any(), any(RequestOptions.class))) + .thenReturn(CompletableFuture.failedFuture(providerRejection)); + try (ExpectedLogEvents capturedLogs = ExpectedLogEvents.capture( + (Logger) LoggerFactory.getLogger(OpenAiCompatibleEmbeddingClient.class)); + OpenAiCompatibleEmbeddingClient embeddingClient = + new OpenAiCompatibleEmbeddingClient(liveClient, batchClient, gatewaySettings())) { + EmbeddingServiceTemporarilyUnavailableException embeddingFailure = assertThrows( + EmbeddingServiceTemporarilyUnavailableException.class, + () -> embeddingClient.embed( + LlmGatewayTier.Purpose.SEARCH_QUERY_EMBEDDING, + List.of("test question"), + LlmGatewayTier.LIVE)); + assertSame(providerRejection, embeddingFailure.getCause()); + assertInstanceOf(RateLimitDecisionException.class, providerRejection.getSuppressed()[0]); + assertTrue(capturedLogs.events().stream().anyMatch(logEvent -> logEvent.getFormattedMessage() + .contains("Embedding provider rejected request"))); + } + } + @Test void batchRetryAfterCooldownDoesNotBlockLiveRequests() { OpenAIClient liveClient = mock(OpenAIClient.class); diff --git a/src/test/java/com/williamcallahan/javachat/web/SseSupportTest.java b/src/test/java/com/williamcallahan/javachat/web/SseSupportTest.java index 2bf966ea..04689624 100644 --- a/src/test/java/com/williamcallahan/javachat/web/SseSupportTest.java +++ b/src/test/java/com/williamcallahan/javachat/web/SseSupportTest.java @@ -12,9 +12,11 @@ import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertTrue; +import ch.qos.logback.classic.Logger; import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.ObjectMapper; import com.williamcallahan.javachat.service.HybridSearchPartialFailureException; +import com.williamcallahan.javachat.support.logging.ExpectedLogEvents; import io.micrometer.core.instrument.simple.SimpleMeterRegistry; import java.time.Duration; import java.util.List; @@ -23,12 +25,15 @@ import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicInteger; import org.junit.jupiter.api.Test; +import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.autoconfigure.json.JsonTest; import org.springframework.http.codec.ServerSentEvent; +import org.springframework.mock.web.MockHttpServletResponse; import reactor.core.Disposable; import reactor.core.Exceptions; import reactor.core.publisher.Flux; +import reactor.core.scheduler.Schedulers; import reactor.test.StepVerifier; import reactor.test.subscriber.TestSubscriber; @@ -48,6 +53,32 @@ class SseSupportTest { @Autowired ObjectMapper objectMapper; + @Test + void retainsHttpRequestCorrelationWhenAnAsynchronousStreamEmitsAnError() { + SseSupport sseSupport = createSseSupport(); + MockHttpServletResponse servletResponse = new MockHttpServletResponse(); + servletResponse.setHeader("X-Request-ID", "stream-correlation-test"); + Flux> failedStream = Flux.defer(() -> sseSupport.configuredProviderUnavailableError()) + .subscribeOn(Schedulers.boundedElastic()); + try (ExpectedLogEvents capturedLogs = + ExpectedLogEvents.capture((Logger) LoggerFactory.getLogger(SseSupport.class))) { + List> emittedEvents = sseSupport + .withRequestDiagnostics(failedStream, servletResponse, "/api/chat/stream") + .collectList() + .block(); + assertEquals( + EVENT_ERROR, + Objects.requireNonNull(emittedEvents).getFirst().event()); + assertEquals(1, capturedLogs.events().size()); + assertTrue(capturedLogs.events().getFirst().getKeyValuePairs().stream() + .anyMatch(logField -> + "requestId".equals(logField.key) && "stream-correlation-test".equals(logField.value))); + assertTrue(capturedLogs.events().getFirst().getKeyValuePairs().stream() + .anyMatch( + logField -> "event".equals(logField.key) && "search_stream_failed".equals(logField.value))); + } + } + @Test void prepareDataStreamPublishesEveryChunkOnce() { SseSupport sseSupport = createSseSupport(); From 4b62161c224c64361f9b0818bb70fb9b079b1a59 Mon Sep 17 00:00:00 2001 From: William Callahan Date: Sun, 1 Nov 2026 16:58:58 -0800 Subject: [PATCH 2/9] fix(search): retain correlated request and failure diagnostics Correlate HTTP and SSE failures and retain browser observations in the existing application logs. Preserve provider 429 causes and retry headers without replaying searches or changing rate limits. Agent-Session: ses_e6699fbbdffeNhL8YBrQk84Urg --- .../in/web/ApiRequestLoggingFilter.java | 55 +++++++------ .../web/BrowserSearchFailureController.java | 15 +++- .../javachat/web/ChatController.java | 3 +- .../web/GuidedLearningController.java | 6 +- .../javachat/web/SseSupport.java | 78 +++++++++++++++---- .../in/web/ApiRequestLoggingFilterTest.java | 47 +++++++++++ .../BrowserSearchFailureControllerTest.java | 37 +++++++++ .../ChatControllerStreamingFailureTest.java | 3 + .../javachat/web/SseSupportTest.java | 49 ++++++++++++ 9 files changed, 251 insertions(+), 42 deletions(-) create mode 100644 src/test/java/com/williamcallahan/javachat/adapters/in/web/ApiRequestLoggingFilterTest.java diff --git a/src/main/java/com/williamcallahan/javachat/adapters/in/web/ApiRequestLoggingFilter.java b/src/main/java/com/williamcallahan/javachat/adapters/in/web/ApiRequestLoggingFilter.java index 676e4299..916d756e 100644 --- a/src/main/java/com/williamcallahan/javachat/adapters/in/web/ApiRequestLoggingFilter.java +++ b/src/main/java/com/williamcallahan/javachat/adapters/in/web/ApiRequestLoggingFilter.java @@ -2,6 +2,7 @@ import com.williamcallahan.javachat.support.StructuredLogValue; import jakarta.servlet.FilterChain; +import jakarta.servlet.RequestDispatcher; import jakarta.servlet.ServletException; import jakarta.servlet.http.HttpServletRequest; import jakarta.servlet.http.HttpServletResponse; @@ -28,7 +29,13 @@ public final class ApiRequestLoggingFilter extends OncePerRequestFilter { /** Excludes static assets and actuator scrapes from application request records. */ @Override protected boolean shouldNotFilter(HttpServletRequest apiRequest) { - return !apiRequest.getRequestURI().startsWith("/api/"); + return !requestPath(apiRequest).startsWith("/api/"); + } + + /** Records an escaping exception only after the container establishes its error response. */ + @Override + protected boolean shouldNotFilterErrorDispatch() { + return false; } /** Includes the completion dispatch rather than timing only servlet setup. */ @@ -43,7 +50,7 @@ protected void doFilterInternal( HttpServletRequest apiRequest, HttpServletResponse apiResponse, FilterChain apiFilterChain) throws ServletException, IOException { String requestId = apiRequest.getRequestId(); - String requestPath = StructuredLogValue.bounded(apiRequest.getRequestURI(), MAX_REQUEST_LOG_FIELD_LENGTH) + String requestPath = StructuredLogValue.bounded(requestPath(apiRequest), MAX_REQUEST_LOG_FIELD_LENGTH) .text(); if (apiRequest.getAttribute(REQUEST_STARTED_ATTRIBUTE) == null) { apiRequest.setAttribute(REQUEST_STARTED_ATTRIBUTE, System.nanoTime()); @@ -54,27 +61,29 @@ protected void doFilterInternal( apiRequest.getMethod(), requestPath); } - try { - apiFilterChain.doFilter(apiRequest, apiResponse); - } finally { - // Spring MVC performs a final ASYNC dispatch after an SSE emitter completes or fails. - if (!apiRequest.isAsyncStarted()) { - long startedNanos = (Long) apiRequest.getAttribute(REQUEST_STARTED_ATTRIBUTE); - log.atLevel(apiResponse.getStatus() >= HttpStatus.BAD_REQUEST.value() ? Level.WARN : Level.INFO) - .setMessage("API request completed") - .addKeyValue("event", "api_request_completed") - .addKeyValue("requestId", requestId) - .addKeyValue("method", apiRequest.getMethod()) - .addKeyValue("path", requestPath) - .addKeyValue("status", apiResponse.getStatus()) - .addKeyValue("durationMs", TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startedNanos)) - .addKeyValue( - "retryAfter", - StructuredLogValue.bounded( - apiResponse.getHeader("Retry-After"), MAX_REQUEST_LOG_FIELD_LENGTH) - .text()) - .log(); - } + apiFilterChain.doFilter(apiRequest, apiResponse); + // Spring MVC performs a final ASYNC dispatch after an SSE emitter completes or fails. + if (!apiRequest.isAsyncStarted()) { + long startedNanos = (Long) apiRequest.getAttribute(REQUEST_STARTED_ATTRIBUTE); + log.atLevel(apiResponse.getStatus() >= HttpStatus.BAD_REQUEST.value() ? Level.WARN : Level.INFO) + .setMessage("API request completed") + .addKeyValue("event", "api_request_completed") + .addKeyValue("requestId", requestId) + .addKeyValue("method", apiRequest.getMethod()) + .addKeyValue("path", requestPath) + .addKeyValue("status", apiResponse.getStatus()) + .addKeyValue("durationMs", TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startedNanos)) + .addKeyValue( + "retryAfter", + StructuredLogValue.bounded( + apiResponse.getHeader("Retry-After"), MAX_REQUEST_LOG_FIELD_LENGTH) + .text()) + .log(); } } + + private static String requestPath(HttpServletRequest apiRequest) { + Object failedPath = apiRequest.getAttribute(RequestDispatcher.ERROR_REQUEST_URI); + return failedPath instanceof String failedRequestPath ? failedRequestPath : apiRequest.getRequestURI(); + } } diff --git a/src/main/java/com/williamcallahan/javachat/adapters/in/web/BrowserSearchFailureController.java b/src/main/java/com/williamcallahan/javachat/adapters/in/web/BrowserSearchFailureController.java index 536e27b0..bc288f48 100644 --- a/src/main/java/com/williamcallahan/javachat/adapters/in/web/BrowserSearchFailureController.java +++ b/src/main/java/com/williamcallahan/javachat/adapters/in/web/BrowserSearchFailureController.java @@ -29,7 +29,10 @@ public ResponseEntity recordFailure(@Valid @RequestBody SearchFailureRepor .setMessage("Browser reported a search failure") .addKeyValue("event", "browser_search_failed") .addKeyValue("provenance", "client_reported") - .addKeyValue("path", failureReport.path()) + .addKeyValue( + "path", + StructuredLogValue.bounded(failureReport.path(), MAX_DIAGNOSTIC_FIELD_LENGTH) + .text()) .addKeyValue("kind", failureReport.kind()) .addKeyValue("httpStatus", failureReport.httpStatus()) .addKeyValue( @@ -46,8 +49,14 @@ public ResponseEntity recordFailure(@Valid @RequestBody SearchFailureRepor .text()) .addKeyValue("retryable", failureReport.retryable()) .addKeyValue("retryAfterSeconds", failureReport.retryAfterSeconds()) - .addKeyValue("pagePath", failureReport.pagePath()) - .addKeyValue("frontendAsset", failureReport.frontendAsset()) + .addKeyValue( + "pagePath", + StructuredLogValue.bounded(failureReport.pagePath(), MAX_DIAGNOSTIC_FIELD_LENGTH) + .text()) + .addKeyValue( + "frontendAsset", + StructuredLogValue.bounded(failureReport.frontendAsset(), MAX_DIAGNOSTIC_FIELD_LENGTH) + .text()) .addKeyValue("elapsedMs", failureReport.elapsedMs()) .log(); return ResponseEntity.accepted().build(); diff --git a/src/main/java/com/williamcallahan/javachat/web/ChatController.java b/src/main/java/com/williamcallahan/javachat/web/ChatController.java index 3efb669d..b81f477b 100644 --- a/src/main/java/com/williamcallahan/javachat/web/ChatController.java +++ b/src/main/java/com/williamcallahan/javachat/web/ChatController.java @@ -280,7 +280,8 @@ public Flux> stream( .log(); } boolean retryable = openAIStreamingService.isRecoverableStreamingFailure(error); - return sseSupport.streamErrorEvent(errorDetail, diagnostics, retryable); + return sseSupport.withFailureDiagnostics( + sseSupport.streamErrorEvent(errorDetail, diagnostics, retryable), error); }), response, "/api/chat/stream"); diff --git a/src/main/java/com/williamcallahan/javachat/web/GuidedLearningController.java b/src/main/java/com/williamcallahan/javachat/web/GuidedLearningController.java index 2850e7e4..8506eedc 100644 --- a/src/main/java/com/williamcallahan/javachat/web/GuidedLearningController.java +++ b/src/main/java/com/williamcallahan/javachat/web/GuidedLearningController.java @@ -427,8 +427,10 @@ private Flux> guidedStreamError( .log(); } boolean retryable = openAIStreamingService.isRecoverableStreamingFailure(error); - return sseSupport.streamErrorEvent( - GUIDED_CHAT_STREAM_ERROR_MESSAGE, GUIDED_CHAT_STREAM_ERROR_DETAILS, retryable); + return sseSupport.withFailureDiagnostics( + sseSupport.streamErrorEvent( + GUIDED_CHAT_STREAM_ERROR_MESSAGE, GUIDED_CHAT_STREAM_ERROR_DETAILS, retryable), + error); } /** diff --git a/src/main/java/com/williamcallahan/javachat/web/SseSupport.java b/src/main/java/com/williamcallahan/javachat/web/SseSupport.java index 88f87743..f9b7ca0b 100644 --- a/src/main/java/com/williamcallahan/javachat/web/SseSupport.java +++ b/src/main/java/com/williamcallahan/javachat/web/SseSupport.java @@ -6,7 +6,9 @@ import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.ObjectWriter; +import com.openai.errors.OpenAIServiceException; import com.williamcallahan.javachat.service.HybridSearchPartialFailureException; +import com.williamcallahan.javachat.support.StructuredLogValue; import io.micrometer.core.instrument.Counter; import io.micrometer.core.instrument.MeterRegistry; import jakarta.servlet.http.HttpServletResponse; @@ -21,6 +23,7 @@ import java.util.function.Consumer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.slf4j.spi.LoggingEventBuilder; import org.springframework.http.HttpHeaders; import org.springframework.http.codec.ServerSentEvent; import org.springframework.stereotype.Component; @@ -39,6 +42,7 @@ @Component public class SseSupport { private static final Logger log = LoggerFactory.getLogger(SseSupport.class); + private static final int MAX_PROVIDER_DIAGNOSTIC_LENGTH = 128; private static final String SOURCE_AVAILABILITY_NOTE = "Note: Matching source documents were not available to JavaChat for this question, so this answer uses the model's general knowledge."; private static final String PREVIOUS_SOURCE_AVAILABILITY_SUFFIX = @@ -136,6 +140,21 @@ public Flux> withRequestDiagnostics( private record StreamRequestObservation(String requestId, String path, long startedNanos) {} + /** Joins the request's terminal event to its own provider refusal, independent of callback threads. */ + public Flux> withFailureDiagnostics( + Flux> errorEvents, Throwable searchFailure) { + Set inspectedFailures = Collections.newSetFromMap(new IdentityHashMap<>()); + Throwable failureCause = searchFailure; + while (failureCause != null && inspectedFailures.add(failureCause)) { + if (failureCause instanceof OpenAIServiceException providerFailure) { + return errorEvents.contextWrite( + streamContext -> streamContext.put(OpenAIServiceException.class, providerFailure)); + } + failureCause = failureCause.getCause(); + } + return errorEvents; + } + /** * Prepares a raw content stream for SSE transmission with backpressure handling. * Filters empty chunks and applies the standard streaming configuration. @@ -248,19 +267,52 @@ public Flux> sseError(SseEventPayload payload) { return Flux.deferContextual(streamContext -> { streamContext .getOrEmpty(StreamRequestObservation.class) - .ifPresent(observation -> log.atWarn() - .setMessage("Search stream emitted a terminal error") - .addKeyValue("event", "search_stream_failed") - .addKeyValue("requestId", observation.requestId()) - .addKeyValue("path", observation.path()) - .addKeyValue("code", payload.code()) - .addKeyValue("stage", payload.stage()) - .addKeyValue("retryable", payload.retryable()) - .addKeyValue( - "durationMs", - java.util.concurrent.TimeUnit.NANOSECONDS.toMillis( - System.nanoTime() - observation.startedNanos())) - .log()); + .ifPresent(observation -> { + LoggingEventBuilder streamFailureLog = log.atWarn() + .setMessage("Search stream emitted a terminal error") + .addKeyValue("event", "search_stream_failed") + .addKeyValue("requestId", observation.requestId()) + .addKeyValue("path", observation.path()) + .addKeyValue("code", payload.code()) + .addKeyValue("stage", payload.stage()) + .addKeyValue("retryable", payload.retryable()) + .addKeyValue( + "durationMs", + java.util.concurrent.TimeUnit.NANOSECONDS.toMillis( + System.nanoTime() - observation.startedNanos())); + streamContext + .getOrEmpty(OpenAIServiceException.class) + .ifPresent(providerFailure -> streamFailureLog + .addKeyValue("upstreamHttpStatus", providerFailure.statusCode()) + .addKeyValue( + "upstreamCode", + StructuredLogValue.bounded( + providerFailure + .code() + .orElse("unknown"), + MAX_PROVIDER_DIAGNOSTIC_LENGTH) + .text()) + .addKeyValue( + "gatewayRequestId", + StructuredLogValue.bounded( + providerFailure + .headers() + .values("x-lgw-request-id") + .stream() + .findFirst() + .orElse("unknown"), + MAX_PROVIDER_DIAGNOSTIC_LENGTH) + .text()) + .addKeyValue( + "retryAfter", + StructuredLogValue.bounded( + providerFailure.headers().values("retry-after").stream() + .findFirst() + .orElse("unknown"), + MAX_PROVIDER_DIAGNOSTIC_LENGTH) + .text())); + streamFailureLog.log(); + }); return Flux.just(errorEvent); }); } diff --git a/src/test/java/com/williamcallahan/javachat/adapters/in/web/ApiRequestLoggingFilterTest.java b/src/test/java/com/williamcallahan/javachat/adapters/in/web/ApiRequestLoggingFilterTest.java new file mode 100644 index 00000000..68762135 --- /dev/null +++ b/src/test/java/com/williamcallahan/javachat/adapters/in/web/ApiRequestLoggingFilterTest.java @@ -0,0 +1,47 @@ +package com.williamcallahan.javachat.adapters.in.web; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import ch.qos.logback.classic.Logger; +import com.williamcallahan.javachat.support.logging.ExpectedLogEvents; +import jakarta.servlet.DispatcherType; +import jakarta.servlet.RequestDispatcher; +import jakarta.servlet.ServletException; +import org.junit.jupiter.api.Test; +import org.slf4j.LoggerFactory; +import org.springframework.mock.web.MockHttpServletRequest; +import org.springframework.mock.web.MockHttpServletResponse; + +/** Verifies that an escaping exception is not recorded as a successful response before error dispatch. */ +class ApiRequestLoggingFilterTest { + @Test + void recordsTheErrorDispatchOutcomeInsteadOfAPrematureSuccess() throws Exception { + ApiRequestLoggingFilter requestLogging = new ApiRequestLoggingFilter(); + MockHttpServletRequest apiRequest = new MockHttpServletRequest("GET", "/api/chat/citations"); + MockHttpServletResponse apiResponse = new MockHttpServletResponse(); + try (ExpectedLogEvents capturedLogs = + ExpectedLogEvents.capture((Logger) LoggerFactory.getLogger(ApiRequestLoggingFilter.class))) { + assertThrows( + ServletException.class, + () -> requestLogging.doFilter(apiRequest, apiResponse, (servletRequest, servletResponse) -> { + throw new ServletException("Retrieval failed"); + })); + assertTrue(capturedLogs.events().stream() + .noneMatch(logEvent -> "API request completed".equals(logEvent.getFormattedMessage()))); + apiRequest.setDispatcherType(DispatcherType.ERROR); + apiRequest.setAttribute(RequestDispatcher.ERROR_REQUEST_URI, "/api/chat/citations"); + apiRequest.setRequestURI("/error"); + apiResponse.setStatus(500); + requestLogging.doFilter(apiRequest, apiResponse, (servletRequest, servletResponse) -> {}); + var completedRequests = capturedLogs.events().stream() + .filter(logEvent -> "API request completed".equals(logEvent.getFormattedMessage())) + .toList(); + assertEquals(1, completedRequests.size()); + assertTrue(completedRequests.getFirst().getKeyValuePairs().stream() + .anyMatch(logField -> "status".equals(logField.key) + && Integer.valueOf(500).equals(logField.value))); + } + } +} diff --git a/src/test/java/com/williamcallahan/javachat/adapters/in/web/BrowserSearchFailureControllerTest.java b/src/test/java/com/williamcallahan/javachat/adapters/in/web/BrowserSearchFailureControllerTest.java index 99d8d3d9..47781c9e 100644 --- a/src/test/java/com/williamcallahan/javachat/adapters/in/web/BrowserSearchFailureControllerTest.java +++ b/src/test/java/com/williamcallahan/javachat/adapters/in/web/BrowserSearchFailureControllerTest.java @@ -1,17 +1,25 @@ package com.williamcallahan.javachat.adapters.in.web; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.springframework.security.test.web.servlet.request.SecurityMockMvcRequestPostProcessors.csrf; import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post; import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; import ch.qos.logback.classic.Logger; +import ch.qos.logback.classic.LoggerContext; +import ch.qos.logback.classic.PatternLayout; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.node.ObjectNode; import com.williamcallahan.javachat.adapters.out.clerk.ClerkApiKeyVerifier; import com.williamcallahan.javachat.config.AppProperties; import com.williamcallahan.javachat.config.SecurityConfig; import com.williamcallahan.javachat.support.logging.ExpectedLogEvents; import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.autoconfigure.web.servlet.WebMvcTest; @@ -34,6 +42,9 @@ class BrowserSearchFailureControllerTest { @Autowired MockMvc mockMvc; + @Autowired + ObjectMapper objectMapper; + @MockitoBean ClerkApiKeyVerifier clerkApiKeyVerifier; @@ -77,4 +88,30 @@ void rejectsQueryStringsAndOversizedDiagnosticFields() throws Exception { .content(REFUSAL_REPORT.replace("diagnostic.probe", "x".repeat(129)))) .andExpect(status().isBadRequest()); } + + @ParameterizedTest + @ValueSource(strings = {"path", "pagePath", "frontendAsset"}) + void browserPathsCannotForgeServerLogFields(String pathField) throws Exception { + ObjectNode reportDocument = assertInstanceOf(ObjectNode.class, objectMapper.readTree(REFUSAL_REPORT)); + reportDocument.put( + pathField, "/api/\" event=\"search_stream_failed\" provenance=\"server\" requestId=\"chosen"); + PatternLayout renderedLog = new PatternLayout(); + renderedLog.setContext((LoggerContext) LoggerFactory.getILoggerFactory()); + renderedLog.setPattern("%kvp"); + renderedLog.start(); + try (ExpectedLogEvents capturedLogs = + ExpectedLogEvents.capture((Logger) LoggerFactory.getLogger(BrowserSearchFailureController.class))) { + mockMvc.perform(post(DIAGNOSTICS_ENDPOINT) + .with(csrf()) + .contentType(MediaType.APPLICATION_JSON) + .content(objectMapper.writeValueAsString(reportDocument))) + .andExpect(status().isAccepted()); + String diagnosticLine = renderedLog.doLayout(capturedLogs.events().getFirst()); + assertFalse(diagnosticLine.contains("event=\"search_stream_failed\"")); + assertFalse(diagnosticLine.contains("provenance=\"server\"")); + assertTrue(diagnosticLine.contains("provenance=\"client_reported\"")); + } finally { + renderedLog.stop(); + } + } } diff --git a/src/test/java/com/williamcallahan/javachat/web/ChatControllerStreamingFailureTest.java b/src/test/java/com/williamcallahan/javachat/web/ChatControllerStreamingFailureTest.java index 45091986..e96dacbb 100644 --- a/src/test/java/com/williamcallahan/javachat/web/ChatControllerStreamingFailureTest.java +++ b/src/test/java/com/williamcallahan/javachat/web/ChatControllerStreamingFailureTest.java @@ -32,6 +32,7 @@ import ch.qos.logback.classic.spi.ILoggingEvent; import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.ObjectMapper; +import com.openai.core.http.Headers; import com.openai.errors.InternalServerException; import com.williamcallahan.javachat.application.streaming.ReportedStreamingFailure; import com.williamcallahan.javachat.config.AppProperties; @@ -118,6 +119,7 @@ void stopCapturingPipelineLogs() { @Test void terminalGatewayTimeoutDoesNotEmitDuplicateControllerError() { InternalServerException upstreamFailure = mock(InternalServerException.class); + when(upstreamFailure.headers()).thenReturn(Headers.builder().build()); when(upstreamFailure.statusCode()).thenReturn(504); when(upstreamFailure.getMessage()).thenReturn(UPSTREAM_SECRET_MESSAGE); ReportedTerminalStreamingFailure terminalFailure = new ReportedTerminalStreamingFailure(upstreamFailure); @@ -245,6 +247,7 @@ void otherUnprocessableEntityFailuresKeepTheGenericMessage() throws JsonProcessi private static ReportedTerminalStreamingFailure unprocessableEntityFailure(String gatewayErrorCode) { InternalServerException gatewayFailure = mock(InternalServerException.class); + when(gatewayFailure.headers()).thenReturn(Headers.builder().build()); when(gatewayFailure.statusCode()).thenReturn(UNPROCESSABLE_ENTITY_STATUS); when(gatewayFailure.code()).thenReturn(Optional.of(gatewayErrorCode)); return new ReportedTerminalStreamingFailure(gatewayFailure); diff --git a/src/test/java/com/williamcallahan/javachat/web/SseSupportTest.java b/src/test/java/com/williamcallahan/javachat/web/SseSupportTest.java index 04689624..690012af 100644 --- a/src/test/java/com/williamcallahan/javachat/web/SseSupportTest.java +++ b/src/test/java/com/williamcallahan/javachat/web/SseSupportTest.java @@ -15,6 +15,9 @@ import ch.qos.logback.classic.Logger; import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.ObjectMapper; +import com.openai.core.http.Headers; +import com.openai.errors.RateLimitException; +import com.williamcallahan.javachat.service.EmbeddingServiceTemporarilyUnavailableException; import com.williamcallahan.javachat.service.HybridSearchPartialFailureException; import com.williamcallahan.javachat.support.logging.ExpectedLogEvents; import io.micrometer.core.instrument.simple.SimpleMeterRegistry; @@ -79,6 +82,52 @@ void retainsHttpRequestCorrelationWhenAnAsynchronousStreamEmitsAnError() { } } + @Test + void concurrentSearchFailuresRetainTheirOwnGatewayRequestAssociation() { + SseSupport sseSupport = createSseSupport(); + Flux> firstFailure = + observedProviderRefusal(sseSupport, "browser-first", "gateway-first"); + Flux> secondFailure = + observedProviderRefusal(sseSupport, "browser-second", "gateway-second"); + try (ExpectedLogEvents capturedLogs = + ExpectedLogEvents.capture((Logger) LoggerFactory.getLogger(SseSupport.class))) { + List> emittedEvents = + Flux.merge(firstFailure, secondFailure).collectList().block(); + assertEquals(2, Objects.requireNonNull(emittedEvents).size()); + for (String requestSuffix : List.of("first", "second")) { + var associatedFailure = capturedLogs.events().stream() + .filter(logEvent -> logEvent.getKeyValuePairs().stream() + .anyMatch(logField -> "requestId".equals(logField.key) + && ("browser-" + requestSuffix).equals(logField.value))) + .findFirst() + .orElseThrow(); + assertTrue(associatedFailure.getKeyValuePairs().stream() + .anyMatch(logField -> "gatewayRequestId".equals(logField.key) + && ("gateway-" + requestSuffix).equals(logField.value))); + assertTrue(associatedFailure.getKeyValuePairs().stream() + .anyMatch(logField -> "upstreamHttpStatus".equals(logField.key) + && Integer.valueOf(429).equals(logField.value))); + } + } + } + + private static Flux> observedProviderRefusal( + SseSupport sseSupport, String browserRequestId, String gatewayRequestId) { + MockHttpServletResponse servletResponse = new MockHttpServletResponse(); + servletResponse.setHeader("X-Request-ID", browserRequestId); + RateLimitException providerRefusal = RateLimitException.builder() + .headers(Headers.builder() + .put("x-lgw-request-id", gatewayRequestId) + .put("Retry-After", "12") + .build()) + .build(); + Flux> failedStream = Flux.defer(() -> sseSupport.withFailureDiagnostics( + sseSupport.configuredProviderUnavailableError(), + new EmbeddingServiceTemporarilyUnavailableException("Provider rate limited", providerRefusal))) + .subscribeOn(Schedulers.boundedElastic()); + return sseSupport.withRequestDiagnostics(failedStream, servletResponse, "/api/chat/stream"); + } + @Test void prepareDataStreamPublishesEveryChunkOnce() { SseSupport sseSupport = createSseSupport(); From 48350277910534190171254d7e9ac30e01769a0a Mon Sep 17 00:00:00 2001 From: William Callahan Date: Sun, 1 Nov 2026 17:07:00 -0800 Subject: [PATCH 3/9] fix(search): retain correlated request and failure diagnostics Correlate HTTP and SSE failures and retain browser observations in the existing application logs. Preserve provider 429 causes and retry headers without replaying searches or changing rate limits. Agent-Session: ses_e6699fbbdffeNhL8YBrQk84Urg --- frontend/src/lib/validation/schemas.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/frontend/src/lib/validation/schemas.ts b/frontend/src/lib/validation/schemas.ts index 4afd1c54..13dbb2dd 100644 --- a/frontend/src/lib/validation/schemas.ts +++ b/frontend/src/lib/validation/schemas.ts @@ -68,7 +68,7 @@ export const StreamErrorSchema = z.object(sseEventFieldShape); export const BrowserSearchFailureSchema = z.object({ path: z.string().startsWith("/api/").max(160), kind: z.enum(["HTTP", "SSE", "NETWORK", "PROTOCOL"]), - httpStatus: z.number().int().min(100).max(599).optional(), + httpStatus: z.int().min(100).max(599).optional(), requestId: z.string().max(128).optional(), code: z.string().max(128).optional(), stage: z.string().max(64).optional(), From 5a6d60db0eeb4a81c3cca271619f082b61e741e5 Mon Sep 17 00:00:00 2001 From: William Callahan Date: Sun, 1 Nov 2026 17:12:39 -0800 Subject: [PATCH 4/9] fix(search): retain correlated request and failure diagnostics Correlate HTTP and SSE failures and retain browser observations in the existing application logs. Preserve provider 429 causes and retry headers without replaying searches or changing rate limits. Agent-Session: ses_e6699fbbdffeNhL8YBrQk84Urg --- .../adapters/in/web/ApiRequestLoggingFilter.java | 15 ++++++++++----- .../williamcallahan/javachat/web/SseSupport.java | 7 ++++--- .../in/web/ApiRequestLoggingFilterTest.java | 3 ++- 3 files changed, 16 insertions(+), 9 deletions(-) diff --git a/src/main/java/com/williamcallahan/javachat/adapters/in/web/ApiRequestLoggingFilter.java b/src/main/java/com/williamcallahan/javachat/adapters/in/web/ApiRequestLoggingFilter.java index 916d756e..f3731088 100644 --- a/src/main/java/com/williamcallahan/javachat/adapters/in/web/ApiRequestLoggingFilter.java +++ b/src/main/java/com/williamcallahan/javachat/adapters/in/web/ApiRequestLoggingFilter.java @@ -55,11 +55,16 @@ protected void doFilterInternal( if (apiRequest.getAttribute(REQUEST_STARTED_ATTRIBUTE) == null) { apiRequest.setAttribute(REQUEST_STARTED_ATTRIBUTE, System.nanoTime()); apiResponse.setHeader(REQUEST_ID_HEADER, requestId); - log.info( - "event=api_request_started requestId={} method={} path={}", - requestId, - apiRequest.getMethod(), - requestPath); + log.atInfo() + .setMessage("API request started") + .addKeyValue("event", "api_request_started") + .addKeyValue("requestId", requestId) + .addKeyValue( + "method", + StructuredLogValue.bounded(apiRequest.getMethod(), MAX_REQUEST_LOG_FIELD_LENGTH) + .text()) + .addKeyValue("path", requestPath) + .log(); } apiFilterChain.doFilter(apiRequest, apiResponse); // Spring MVC performs a final ASYNC dispatch after an SSE emitter completes or fails. diff --git a/src/main/java/com/williamcallahan/javachat/web/SseSupport.java b/src/main/java/com/williamcallahan/javachat/web/SseSupport.java index f9b7ca0b..2f1f048a 100644 --- a/src/main/java/com/williamcallahan/javachat/web/SseSupport.java +++ b/src/main/java/com/williamcallahan/javachat/web/SseSupport.java @@ -282,7 +282,7 @@ public Flux> sseError(SseEventPayload payload) { System.nanoTime() - observation.startedNanos())); streamContext .getOrEmpty(OpenAIServiceException.class) - .ifPresent(providerFailure -> streamFailureLog + .map(providerFailure -> streamFailureLog .addKeyValue("upstreamHttpStatus", providerFailure.statusCode()) .addKeyValue( "upstreamCode", @@ -310,8 +310,9 @@ public Flux> sseError(SseEventPayload payload) { .findFirst() .orElse("unknown"), MAX_PROVIDER_DIAGNOSTIC_LENGTH) - .text())); - streamFailureLog.log(); + .text())) + .orElse(streamFailureLog) + .log(); }); return Flux.just(errorEvent); }); diff --git a/src/test/java/com/williamcallahan/javachat/adapters/in/web/ApiRequestLoggingFilterTest.java b/src/test/java/com/williamcallahan/javachat/adapters/in/web/ApiRequestLoggingFilterTest.java index 68762135..e3a8b72d 100644 --- a/src/test/java/com/williamcallahan/javachat/adapters/in/web/ApiRequestLoggingFilterTest.java +++ b/src/test/java/com/williamcallahan/javachat/adapters/in/web/ApiRequestLoggingFilterTest.java @@ -9,6 +9,7 @@ import jakarta.servlet.DispatcherType; import jakarta.servlet.RequestDispatcher; import jakarta.servlet.ServletException; +import java.io.IOException; import org.junit.jupiter.api.Test; import org.slf4j.LoggerFactory; import org.springframework.mock.web.MockHttpServletRequest; @@ -17,7 +18,7 @@ /** Verifies that an escaping exception is not recorded as a successful response before error dispatch. */ class ApiRequestLoggingFilterTest { @Test - void recordsTheErrorDispatchOutcomeInsteadOfAPrematureSuccess() throws Exception { + void recordsTheErrorDispatchOutcomeInsteadOfAPrematureSuccess() throws ServletException, IOException { ApiRequestLoggingFilter requestLogging = new ApiRequestLoggingFilter(); MockHttpServletRequest apiRequest = new MockHttpServletRequest("GET", "/api/chat/citations"); MockHttpServletResponse apiResponse = new MockHttpServletResponse(); From e8ad2ce6a06ae0d8efc2f996e466aed5ce4dc0c0 Mon Sep 17 00:00:00 2001 From: William Callahan Date: Sun, 1 Nov 2026 17:23:28 -0800 Subject: [PATCH 5/9] fix(search): retain correlated request and failure diagnostics Correlate HTTP and SSE failures and retain browser observations in the existing application logs. Preserve provider 429 causes and retry headers without replaying searches or changing rate limits. Agent-Session: ses_e6699fbbdffeNhL8YBrQk84Urg --- frontend/e2e/search-diagnostics.spec.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/frontend/e2e/search-diagnostics.spec.ts b/frontend/e2e/search-diagnostics.spec.ts index 8b234ed6..1d9b3797 100644 --- a/frontend/e2e/search-diagnostics.spec.ts +++ b/frontend/e2e/search-diagnostics.spec.ts @@ -27,7 +27,7 @@ for (const failureKind of ["HTTP", "SSE"] as const) { await page.getByLabel("Message input").fill("Explain Java records"); await page.getByLabel("Message input").press("Enter"); await expect( - page.getByText("Search is temporarily rate limited", { exact: true }), + page.getByRole("status").getByText("Search is temporarily rate limited", { exact: true }), ).toBeVisible(); const diagnosticRequest = await diagnosticSubmission; const parsedDiagnostic: unknown = diagnosticRequest.postDataJSON(); From 4cf3e383410e8ad9698b9edd51cbb2b256a9fb36 Mon Sep 17 00:00:00 2001 From: William Callahan Date: Sun, 1 Nov 2026 17:48:24 -0800 Subject: [PATCH 6/9] fix(search): retain correlated request and failure diagnostics Correlate HTTP and SSE failures and retain browser observations in the existing application logs. Preserve provider 429 causes and retry headers without replaying searches or changing rate limits. Agent-Session: ses_e6699fbbdffeNhL8YBrQk84Urg --- docs/api.md | 2 + .../OpenAiStreamingFailureException.java | 13 ++++++ .../streaming/ReportedStreamingFailure.java | 5 +++ .../streaming/StreamingFailureReporter.java | 12 +++++- .../service/OpenAIStreamingService.java | 7 +++- .../javachat/web/SseSupport.java | 15 ++++++- .../OpenAiStreamingFailureExceptionTest.java | 22 +++++++--- .../service/OpenAIStreamingServiceTest.java | 6 ++- .../javachat/web/SseSupportTest.java | 40 +++++++++++++++++++ 9 files changed, 111 insertions(+), 11 deletions(-) diff --git a/docs/api.md b/docs/api.md index 7eb5536a..85ed02b6 100644 --- a/docs/api.md +++ b/docs/api.md @@ -55,6 +55,8 @@ Application logs distinguish HTTP completion (`api_request_completed`) from a te error (`search_stream_failed`); HTTP 200 alone does not establish that the answer succeeded. Embedding rate-limit records retain the upstream status, workload, model, gateway request ID, and `Retry-After` when supplied. Missing retry timing does not replace the original provider cause. +An in-band Responses failure retains its finite `terminalReason` on the request-correlated log; +that protocol outcome is not reported as a transport HTTP 429. ### Browser failure diagnostics diff --git a/src/main/java/com/williamcallahan/javachat/adapters/out/llm/openai/OpenAiStreamingFailureException.java b/src/main/java/com/williamcallahan/javachat/adapters/out/llm/openai/OpenAiStreamingFailureException.java index f2e8049e..05509bea 100644 --- a/src/main/java/com/williamcallahan/javachat/adapters/out/llm/openai/OpenAiStreamingFailureException.java +++ b/src/main/java/com/williamcallahan/javachat/adapters/out/llm/openai/OpenAiStreamingFailureException.java @@ -46,6 +46,7 @@ enum AlertField { UPSTREAM_TYPE("upstreamType"), GATEWAY_REQUEST_ID("gatewayRequestId"), RETRY_AFTER("retryAfter"), + TERMINAL_REASON("terminalReason"), QUEUE_TIER("queueTier"), QUEUE_LANE("queueLane"), QUEUE_TIMEOUT_PHASE("queueTimeoutPhase"), @@ -117,6 +118,7 @@ String wireName() { private final String upstreamType; private final String gatewayRequestId; private final String retryAfter; + private final String terminalReason; private final Map queueHeaderValues; private OpenAiStreamingFailureException(Throwable upstreamFailure, TerminalAttempt terminalAttempt) { @@ -129,6 +131,8 @@ private OpenAiStreamingFailureException(Throwable upstreamFailure, TerminalAttem this.currentAttempt = validatedAttempt.currentAttempt(); this.maxAttempts = validatedAttempt.maxAttempts(); this.emittedTextChunk = validatedAttempt.emittedTextChunk(); + this.terminalReason = + boundedOptionalValue(validatedAttempt.terminalReason()).orElse(null); if (upstreamFailure instanceof OpenAIServiceException serviceException) { this.upstreamHttpStatus = serviceException.statusCode(); this.upstreamCode = boundedOptionalValue(serviceException.code()).orElse(null); @@ -172,6 +176,12 @@ public Throwable upstreamFailure() { return Objects.requireNonNull(getCause(), "upstreamFailure"); } + /** Exposes the finite in-band reason to the request-correlated terminal log. */ + @Override + public Optional terminalReason() { + return Optional.ofNullable(terminalReason); + } + /** Returns the bounded provider name used by the terminal attempt. */ String providerName() { return providerName; @@ -270,6 +280,9 @@ private void logTerminalAlert() { if (retryAfter != null) { terminalAlert = terminalAlert.addKeyValue(AlertField.RETRY_AFTER.fieldName(), retryAfter); } + if (terminalReason != null) { + terminalAlert = terminalAlert.addKeyValue(AlertField.TERMINAL_REASON.fieldName(), terminalReason); + } for (QueueHeader queueHeader : QueueHeader.values()) { if (queueHeaderValues.containsKey(queueHeader)) { terminalAlert = terminalAlert.addKeyValue( diff --git a/src/main/java/com/williamcallahan/javachat/application/streaming/ReportedStreamingFailure.java b/src/main/java/com/williamcallahan/javachat/application/streaming/ReportedStreamingFailure.java index 92f66192..680e9a4b 100644 --- a/src/main/java/com/williamcallahan/javachat/application/streaming/ReportedStreamingFailure.java +++ b/src/main/java/com/williamcallahan/javachat/application/streaming/ReportedStreamingFailure.java @@ -21,6 +21,11 @@ public interface ReportedStreamingFailure { */ Throwable upstreamFailure(); + /** Returns a retained finite protocol reason without treating an in-band failure as an HTTP refusal. */ + default Optional terminalReason() { + return Optional.empty(); + } + /** * Finds the first reported streaming failure in a cause chain without revisiting identity cycles. * diff --git a/src/main/java/com/williamcallahan/javachat/application/streaming/StreamingFailureReporter.java b/src/main/java/com/williamcallahan/javachat/application/streaming/StreamingFailureReporter.java index f45d608f..4663f7c8 100644 --- a/src/main/java/com/williamcallahan/javachat/application/streaming/StreamingFailureReporter.java +++ b/src/main/java/com/williamcallahan/javachat/application/streaming/StreamingFailureReporter.java @@ -1,5 +1,8 @@ package com.williamcallahan.javachat.application.streaming; +import java.util.Objects; +import java.util.Optional; + /** * Reports a terminal provider failure without exposing provider-specific exception types to orchestration code. */ @@ -16,7 +19,12 @@ public interface StreamingFailureReporter { /** Captures the provider attempt that became terminal. */ record TerminalAttempt( - String providerName, String modelId, int currentAttempt, int maxAttempts, boolean emittedTextChunk) { + String providerName, + String modelId, + int currentAttempt, + int maxAttempts, + boolean emittedTextChunk, + Optional terminalReason) { /** * Validates immutable context before a provider reporter can emit it. @@ -26,8 +34,10 @@ record TerminalAttempt( * @param currentAttempt one-based terminal attempt number * @param maxAttempts maximum number of allowed attempts * @param emittedTextChunk whether visible text was emitted before the failure + * @param terminalReason finite protocol reason for an in-band failure; empty for transport failures */ public TerminalAttempt { + terminalReason = Objects.requireNonNull(terminalReason, "terminalReason"); if (providerName == null || providerName.isBlank()) { throw new IllegalArgumentException("providerName must not be blank"); } diff --git a/src/main/java/com/williamcallahan/javachat/service/OpenAIStreamingService.java b/src/main/java/com/williamcallahan/javachat/service/OpenAIStreamingService.java index 5cc0fd91..3a9119b5 100644 --- a/src/main/java/com/williamcallahan/javachat/service/OpenAIStreamingService.java +++ b/src/main/java/com/williamcallahan/javachat/service/OpenAIStreamingService.java @@ -289,7 +289,12 @@ private Flux executeStreamingWithConfiguredProvider( preparedStreamingRequest.modelId(), 1, 1, - emittedVisibleText.get()))); + emittedVisibleText.get(), + streamingFailure instanceof OpenAiResponseException responseFailure + ? Optional.of(responseFailure + .terminalReason() + .name()) + : Optional.empty()))); }); }); } diff --git a/src/main/java/com/williamcallahan/javachat/web/SseSupport.java b/src/main/java/com/williamcallahan/javachat/web/SseSupport.java index 2f1f048a..1ba7e956 100644 --- a/src/main/java/com/williamcallahan/javachat/web/SseSupport.java +++ b/src/main/java/com/williamcallahan/javachat/web/SseSupport.java @@ -7,6 +7,7 @@ import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.ObjectWriter; import com.openai.errors.OpenAIServiceException; +import com.williamcallahan.javachat.application.streaming.ReportedStreamingFailure; import com.williamcallahan.javachat.service.HybridSearchPartialFailureException; import com.williamcallahan.javachat.support.StructuredLogValue; import io.micrometer.core.instrument.Counter; @@ -143,16 +144,20 @@ private record StreamRequestObservation(String requestId, String path, long star /** Joins the request's terminal event to its own provider refusal, independent of callback threads. */ public Flux> withFailureDiagnostics( Flux> errorEvents, Throwable searchFailure) { + Flux> correlatedErrors = ReportedStreamingFailure.findInCauseChain(searchFailure) + .map(reportedFailure -> errorEvents.contextWrite( + streamContext -> streamContext.put(ReportedStreamingFailure.class, reportedFailure))) + .orElse(errorEvents); Set inspectedFailures = Collections.newSetFromMap(new IdentityHashMap<>()); Throwable failureCause = searchFailure; while (failureCause != null && inspectedFailures.add(failureCause)) { if (failureCause instanceof OpenAIServiceException providerFailure) { - return errorEvents.contextWrite( + return correlatedErrors.contextWrite( streamContext -> streamContext.put(OpenAIServiceException.class, providerFailure)); } failureCause = failureCause.getCause(); } - return errorEvents; + return correlatedErrors; } /** @@ -276,6 +281,12 @@ public Flux> sseError(SseEventPayload payload) { .addKeyValue("code", payload.code()) .addKeyValue("stage", payload.stage()) .addKeyValue("retryable", payload.retryable()) + .addKeyValue( + "terminalReason", + streamContext + .getOrEmpty(ReportedStreamingFailure.class) + .flatMap(ReportedStreamingFailure::terminalReason) + .orElse(null)) .addKeyValue( "durationMs", java.util.concurrent.TimeUnit.NANOSECONDS.toMillis( diff --git a/src/test/java/com/williamcallahan/javachat/adapters/out/llm/openai/OpenAiStreamingFailureExceptionTest.java b/src/test/java/com/williamcallahan/javachat/adapters/out/llm/openai/OpenAiStreamingFailureExceptionTest.java index 6d1ccf4b..4e5d5770 100644 --- a/src/test/java/com/williamcallahan/javachat/adapters/out/llm/openai/OpenAiStreamingFailureExceptionTest.java +++ b/src/test/java/com/williamcallahan/javachat/adapters/out/llm/openai/OpenAiStreamingFailureExceptionTest.java @@ -191,15 +191,25 @@ void reportedFailureTraversalStopsAtIdentityCycleWithoutMarker() { @Test void terminalAttemptRejectsInvalidProviderModelAndAttemptBounds() { - assertThrows(IllegalArgumentException.class, () -> new TerminalAttempt(PROVIDER_NAME, MODEL_ID, 0, 1, false)); - assertThrows(IllegalArgumentException.class, () -> new TerminalAttempt(PROVIDER_NAME, MODEL_ID, 2, 1, false)); - assertThrows(IllegalArgumentException.class, () -> new TerminalAttempt(" ", MODEL_ID, 1, 1, false)); - assertThrows(IllegalArgumentException.class, () -> new TerminalAttempt(PROVIDER_NAME, "", 1, 1, false)); - assertThrows(IllegalArgumentException.class, () -> new TerminalAttempt(PROVIDER_NAME, MODEL_ID, 1, 0, false)); + assertThrows( + IllegalArgumentException.class, + () -> new TerminalAttempt(PROVIDER_NAME, MODEL_ID, 0, 1, false, java.util.Optional.empty())); + assertThrows( + IllegalArgumentException.class, + () -> new TerminalAttempt(PROVIDER_NAME, MODEL_ID, 2, 1, false, java.util.Optional.empty())); + assertThrows( + IllegalArgumentException.class, + () -> new TerminalAttempt(" ", MODEL_ID, 1, 1, false, java.util.Optional.empty())); + assertThrows( + IllegalArgumentException.class, + () -> new TerminalAttempt(PROVIDER_NAME, "", 1, 1, false, java.util.Optional.empty())); + assertThrows( + IllegalArgumentException.class, + () -> new TerminalAttempt(PROVIDER_NAME, MODEL_ID, 1, 0, false, java.util.Optional.empty())); } private static TerminalAttempt terminalAttempt(boolean emittedTextChunk) { - return new TerminalAttempt(PROVIDER_NAME, MODEL_ID, 2, 2, emittedTextChunk); + return new TerminalAttempt(PROVIDER_NAME, MODEL_ID, 2, 2, emittedTextChunk, java.util.Optional.empty()); } private ILoggingEvent onlyTerminalAlert() { diff --git a/src/test/java/com/williamcallahan/javachat/service/OpenAIStreamingServiceTest.java b/src/test/java/com/williamcallahan/javachat/service/OpenAIStreamingServiceTest.java index 4fbde71d..dddfb80b 100644 --- a/src/test/java/com/williamcallahan/javachat/service/OpenAIStreamingServiceTest.java +++ b/src/test/java/com/williamcallahan/javachat/service/OpenAIStreamingServiceTest.java @@ -122,7 +122,7 @@ void recoverableStreamingFailureUnwrapsNestedTerminalContext() { OpenAiStreamingFailureException terminalFailure = OpenAiStreamingFailureException.terminalAndLog( internalServerException, new StreamingFailureReporter.TerminalAttempt( - RateLimitService.ApiProvider.OPENAI.getName(), "gpt-5.4", 1, 1, false)); + RateLimitService.ApiProvider.OPENAI.getName(), "gpt-5.4", 1, 1, false, Optional.empty())); assertTrue(streamingService.isRecoverableStreamingFailure( new IllegalStateException("reactor boundary", terminalFailure))); @@ -432,6 +432,10 @@ void failedRateLimitResponseStartsBackoffWithoutImmediateRetry() { OpenAiResponseException.TerminalReason.RATE_LIMIT_EXCEEDED, upstreamFailure.terminalReason()); assertFalse(streamingService.isRecoverableStreamingFailure(failure)); + assertEquals( + Optional.of("RATE_LIMIT_EXCEEDED"), + assertInstanceOf(OpenAiStreamingFailureException.class, failure) + .terminalReason()); }) .verify(); diff --git a/src/test/java/com/williamcallahan/javachat/web/SseSupportTest.java b/src/test/java/com/williamcallahan/javachat/web/SseSupportTest.java index 690012af..3ebcea75 100644 --- a/src/test/java/com/williamcallahan/javachat/web/SseSupportTest.java +++ b/src/test/java/com/williamcallahan/javachat/web/SseSupportTest.java @@ -17,6 +17,8 @@ import com.fasterxml.jackson.databind.ObjectMapper; import com.openai.core.http.Headers; import com.openai.errors.RateLimitException; +import com.williamcallahan.javachat.adapters.out.llm.openai.OpenAiStreamingFailureException; +import com.williamcallahan.javachat.application.streaming.StreamingFailureReporter.TerminalAttempt; import com.williamcallahan.javachat.service.EmbeddingServiceTemporarilyUnavailableException; import com.williamcallahan.javachat.service.HybridSearchPartialFailureException; import com.williamcallahan.javachat.support.logging.ExpectedLogEvents; @@ -24,6 +26,7 @@ import java.time.Duration; import java.util.List; import java.util.Objects; +import java.util.Optional; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicInteger; @@ -111,6 +114,43 @@ void concurrentSearchFailuresRetainTheirOwnGatewayRequestAssociation() { } } + @Test + void inBandRateLimitRetainsItsReasonAndRequestWithoutInventingAnHttpStatus() { + SseSupport sseSupport = createSseSupport(); + MockHttpServletResponse servletResponse = new MockHttpServletResponse(); + servletResponse.setHeader("X-Request-ID", "in-band-request"); + try (ExpectedLogEvents providerLogs = ExpectedLogEvents.capture( + (Logger) LoggerFactory.getLogger(OpenAiStreamingFailureException.class)); + ExpectedLogEvents requestLogs = + ExpectedLogEvents.capture((Logger) LoggerFactory.getLogger(SseSupport.class))) { + OpenAiStreamingFailureException reportedFailure = OpenAiStreamingFailureException.terminalAndLog( + new IllegalStateException("Controlled in-band failure"), + new TerminalAttempt("openai", "test-model", 1, 1, false, Optional.of("RATE_LIMIT_EXCEEDED"))); + var failedEvents = sseSupport.withFailureDiagnostics( + sseSupport.streamErrorEvent("Failure", "In-band refusal", false), reportedFailure); + var emittedEvents = sseSupport + .withRequestDiagnostics(failedEvents, servletResponse, "/api/chat/stream") + .collectList() + .block(); + assertEquals( + EVENT_ERROR, + Objects.requireNonNull(emittedEvents).getFirst().event()); + var correlatedFailure = requestLogs.events().getFirst(); + assertTrue(correlatedFailure.getKeyValuePairs().stream() + .anyMatch(logField -> + "terminalReason".equals(logField.key) && "RATE_LIMIT_EXCEEDED".equals(logField.value))); + assertTrue(correlatedFailure.getKeyValuePairs().stream() + .anyMatch( + logField -> "requestId".equals(logField.key) && "in-band-request".equals(logField.value))); + assertFalse(correlatedFailure.getKeyValuePairs().stream() + .anyMatch(logField -> "upstreamHttpStatus".equals(logField.key))); + assertEquals(1, providerLogs.events().size()); + assertTrue(providerLogs.events().getFirst().getKeyValuePairs().stream() + .anyMatch(logField -> + "terminalReason".equals(logField.key) && "RATE_LIMIT_EXCEEDED".equals(logField.value))); + } + } + private static Flux> observedProviderRefusal( SseSupport sseSupport, String browserRequestId, String gatewayRequestId) { MockHttpServletResponse servletResponse = new MockHttpServletResponse(); From 9a23ae68c38afa43728fee841530941a2a12d1b6 Mon Sep 17 00:00:00 2001 From: William Callahan Date: Sun, 1 Nov 2026 22:26:49 -0800 Subject: [PATCH 7/9] fix(diagnostics): report final request and stream failure outcomes Defer sendError completion until error rendering and retain one completion record across dispatches. Emit stream diagnostics only with the terminal event, preserve protocol failures and bounded browser observations, and assert the persistent browser error message. Agent-Session: ses_e651faff3ffeDdTophrwg3Q2V7 --- frontend/e2e/search-diagnostics.spec.ts | 4 +- .../lib/services/searchDiagnostics.test.ts | 68 +++++++++- frontend/src/lib/services/sse.ts | 5 +- .../in/web/ApiRequestLoggingFilter.java | 46 ++++++- .../javachat/web/SseSupport.java | 25 ++-- .../in/web/ApiRequestLoggingFilterTest.java | 124 +++++++++++++++++- .../javachat/config/SecurityConfigTest.java | 49 ++++++- .../javachat/web/SseSupportTest.java | 30 +++++ 8 files changed, 327 insertions(+), 24 deletions(-) diff --git a/frontend/e2e/search-diagnostics.spec.ts b/frontend/e2e/search-diagnostics.spec.ts index 1d9b3797..047c3895 100644 --- a/frontend/e2e/search-diagnostics.spec.ts +++ b/frontend/e2e/search-diagnostics.spec.ts @@ -27,7 +27,9 @@ for (const failureKind of ["HTTP", "SSE"] as const) { await page.getByLabel("Message input").fill("Explain Java records"); await page.getByLabel("Message input").press("Enter"); await expect( - page.getByRole("status").getByText("Search is temporarily rate limited", { exact: true }), + page + .locator(".message.assistant.error .bubble") + .getByText("Search is temporarily rate limited", { exact: true }), ).toBeVisible(); const diagnosticRequest = await diagnosticSubmission; const parsedDiagnostic: unknown = diagnosticRequest.postDataJSON(); diff --git a/frontend/src/lib/services/searchDiagnostics.test.ts b/frontend/src/lib/services/searchDiagnostics.test.ts index 86824229..4c8d35a0 100644 --- a/frontend/src/lib/services/searchDiagnostics.test.ts +++ b/frontend/src/lib/services/searchDiagnostics.test.ts @@ -1,14 +1,17 @@ import { afterEach, describe, expect, it, vi } from "vitest"; import { BrowserSearchFailureSchema } from "../validation/schemas"; -import { streamSse } from "./sse"; +import { streamSse, streamSseGet } from "./sse"; describe("search failure diagnostic transport", () => { afterEach(() => { + window.history.replaceState({}, "", "/"); vi.unstubAllGlobals(); vi.restoreAllMocks(); }); - it("retains one HTTP refusal report without replaying the search or transmitting its prompt", async () => { + it.each([1, 160, 161])("retains HTTP diagnostics at page length %i", async (pagePathLength) => { + const pagePath = `/${"a".repeat(pagePathLength - 1)}`; + window.history.replaceState({}, "", `${pagePath}?question=private-query#private-fragment`); const submittedReports: string[] = []; const searchRequests: string[] = []; vi.stubGlobal( @@ -42,6 +45,8 @@ describe("search failure diagnostic transport", () => { await vi.waitFor(() => expect(submittedReports).toHaveLength(1)); expect(searchRequests).toEqual(["/api/chat/stream"]); expect(submittedReports[0]).not.toContain("private search question"); + expect(submittedReports[0]).not.toContain("private-query"); + expect(submittedReports[0]).not.toContain("private-fragment"); const diagnostic = BrowserSearchFailureSchema.safeParse( JSON.parse(submittedReports[0] ?? "null"), ); @@ -52,9 +57,68 @@ describe("search failure diagnostic transport", () => { httpStatus: 429, requestId: "refusal-report-test", retryAfterSeconds: 12, + pagePath: `/${"a".repeat(Math.min(pagePathLength, 160) - 1)}`, }); }); + it.each([ + { method: "GET", path: "/api/guided/content/stream" }, + { method: "POST", path: "/api/chat/stream" }, + ])( + "reports a missing $method response body as PROTOCOL without replaying the search", + async ({ method, path }) => { + const submittedReports: string[] = []; + const searchRequests: string[] = []; + const searchUrl = `${path}?slug=java-records`; + vi.stubGlobal( + "fetch", + vi.fn(async (url: string, options?: RequestInit) => { + if (url === "/api/diagnostics/search-failures") { + if (typeof options?.body !== "string") + throw new Error("Expected a JSON diagnostic request body"); + submittedReports.push(options.body); + return new Response(null, { status: 202 }); + } + searchRequests.push(url); + return new Response(null, { + status: 200, + headers: { "X-Request-ID": "missing-body-report-test" }, + }); + }), + ); + + const streamingRequest = + method === "GET" + ? streamSseGet(searchUrl, { onText: vi.fn() }, "guided.lesson-content.ts") + : streamSse( + searchUrl, + { latest: "private search question" }, + { onText: vi.fn() }, + "chat.ts", + ); + await expect(streamingRequest).rejects.toMatchObject({ + name: "StreamFailureError", + message: "No response body", + retryable: true, + }); + await vi.waitFor(() => expect(submittedReports).toHaveLength(1)); + expect(searchRequests).toEqual([searchUrl]); + expect(submittedReports[0]).not.toContain("private search question"); + expect(submittedReports[0]).not.toContain("java-records"); + const diagnostic = BrowserSearchFailureSchema.safeParse( + JSON.parse(submittedReports[0] ?? "null"), + ); + if (!diagnostic.success) throw diagnostic.error; + expect(diagnostic.data).toMatchObject({ + kind: "PROTOCOL", + path, + httpStatus: 200, + requestId: "missing-body-report-test", + retryable: true, + }); + }, + ); + it("distinguishes an SSE error inside HTTP 200 from an HTTP refusal", async () => { const submittedReports: string[] = []; vi.stubGlobal( diff --git a/frontend/src/lib/services/sse.ts b/frontend/src/lib/services/sse.ts index 2e8bb58c..446708da 100644 --- a/frontend/src/lib/services/sse.ts +++ b/frontend/src/lib/services/sse.ts @@ -29,6 +29,7 @@ const SSE_EVENT_CITATION = "citation"; const SSE_EVENT_PROVIDER = "provider"; const SSE_EVENT_TEXT = "text"; const INVALID_SSE_EVENT_MESSAGE = "Received an invalid SSE event from the server"; +const DIAGNOSTIC_PAGE_PATH_MAX_LENGTH = 160; /** Optional request options for streaming fetch calls. */ export interface StreamSseRequestOptions { @@ -339,7 +340,7 @@ function reportStreamFailure( failureKind = "HTTP"; } else if (streamFailure?.cause) { failureKind = "NETWORK"; - } else if (streamFailure) { + } else if (streamFailure && httpResponse.body !== null) { failureKind = "SSE"; } void reportSearchFailure({ @@ -351,7 +352,7 @@ function reportStreamFailure( stage: streamFailure?.stage, retryable: streamFailure?.retryable, retryAfterSeconds: streamFailure?.retryAfterSeconds, - pagePath: window.location.pathname, + pagePath: window.location.pathname.slice(0, DIAGNOSTIC_PAGE_PATH_MAX_LENGTH), frontendAsset: frontendScript ? new URL(frontendScript.src).pathname : undefined, elapsedMs: Math.round(performance.now() - startedAt), }); diff --git a/src/main/java/com/williamcallahan/javachat/adapters/in/web/ApiRequestLoggingFilter.java b/src/main/java/com/williamcallahan/javachat/adapters/in/web/ApiRequestLoggingFilter.java index f3731088..220b504b 100644 --- a/src/main/java/com/williamcallahan/javachat/adapters/in/web/ApiRequestLoggingFilter.java +++ b/src/main/java/com/williamcallahan/javachat/adapters/in/web/ApiRequestLoggingFilter.java @@ -1,11 +1,13 @@ package com.williamcallahan.javachat.adapters.in.web; import com.williamcallahan.javachat.support.StructuredLogValue; +import jakarta.servlet.DispatcherType; import jakarta.servlet.FilterChain; import jakarta.servlet.RequestDispatcher; import jakarta.servlet.ServletException; import jakarta.servlet.http.HttpServletRequest; import jakarta.servlet.http.HttpServletResponse; +import jakarta.servlet.http.HttpServletResponseWrapper; import java.io.IOException; import java.util.concurrent.TimeUnit; import org.slf4j.Logger; @@ -23,6 +25,8 @@ public final class ApiRequestLoggingFilter extends OncePerRequestFilter { private static final Logger log = LoggerFactory.getLogger(ApiRequestLoggingFilter.class); private static final String REQUEST_STARTED_ATTRIBUTE = ApiRequestLoggingFilter.class.getName() + ".started"; + private static final String REQUEST_COMPLETED_ATTRIBUTE = ApiRequestLoggingFilter.class.getName() + ".completed"; + private static final String ERROR_PENDING_ATTRIBUTE = ApiRequestLoggingFilter.class.getName() + ".errorPending"; private static final String REQUEST_ID_HEADER = "X-Request-ID"; private static final int MAX_REQUEST_LOG_FIELD_LENGTH = 256; @@ -66,9 +70,14 @@ protected void doFilterInternal( .addKeyValue("path", requestPath) .log(); } - apiFilterChain.doFilter(apiRequest, apiResponse); + boolean errorDispatch = apiRequest.getDispatcherType() == DispatcherType.ERROR; + apiFilterChain.doFilter(apiRequest, new ErrorDispatchResponse(apiRequest, apiResponse)); + // sendError returns before the container renders its error page, including security denials. // Spring MVC performs a final ASYNC dispatch after an SSE emitter completes or fails. - if (!apiRequest.isAsyncStarted()) { + if (!apiRequest.isAsyncStarted() + && (errorDispatch || apiRequest.getAttribute(ERROR_PENDING_ATTRIBUTE) == null) + && apiRequest.getAttribute(REQUEST_COMPLETED_ATTRIBUTE) == null) { + apiRequest.setAttribute(REQUEST_COMPLETED_ATTRIBUTE, Boolean.TRUE); long startedNanos = (Long) apiRequest.getAttribute(REQUEST_STARTED_ATTRIBUTE); log.atLevel(apiResponse.getStatus() >= HttpStatus.BAD_REQUEST.value() ? Level.WARN : Level.INFO) .setMessage("API request completed") @@ -87,6 +96,39 @@ protected void doFilterInternal( } } + /** Retains the same completion ownership when the container nests its error page dispatch. */ + @Override + protected void doFilterNestedErrorDispatch( + HttpServletRequest apiRequest, HttpServletResponse apiResponse, FilterChain apiFilterChain) + throws ServletException, IOException { + apiFilterChain.doFilter(apiRequest, apiResponse); + // The enclosing dispatch owns completion, after every outer filter has returned. + apiRequest.removeAttribute(ERROR_PENDING_ATTRIBUTE); + } + + private static final class ErrorDispatchResponse extends HttpServletResponseWrapper { + private final HttpServletRequest apiRequest; + + private ErrorDispatchResponse(HttpServletRequest apiRequest, HttpServletResponse apiResponse) { + super(apiResponse); + this.apiRequest = apiRequest; + } + + /** Defers completion until the container renders the error response. */ + @Override + public void sendError(int status) throws IOException { + apiRequest.setAttribute(ERROR_PENDING_ATTRIBUTE, Boolean.TRUE); + super.sendError(status); + } + + /** Defers completion for error responses carrying a container-rendered message. */ + @Override + public void sendError(int status, String message) throws IOException { + apiRequest.setAttribute(ERROR_PENDING_ATTRIBUTE, Boolean.TRUE); + super.sendError(status, message); + } + } + private static String requestPath(HttpServletRequest apiRequest) { Object failedPath = apiRequest.getAttribute(RequestDispatcher.ERROR_REQUEST_URI); return failedPath instanceof String failedRequestPath ? failedRequestPath : apiRequest.getRequestURI(); diff --git a/src/main/java/com/williamcallahan/javachat/web/SseSupport.java b/src/main/java/com/williamcallahan/javachat/web/SseSupport.java index 1ba7e956..7ca219e4 100644 --- a/src/main/java/com/williamcallahan/javachat/web/SseSupport.java +++ b/src/main/java/com/williamcallahan/javachat/web/SseSupport.java @@ -259,17 +259,19 @@ public Flux> sseError(String message, String details) { * Uses a localized fallback when error serialization itself fails, since this is the * terminal error path with nowhere further to propagate. */ - public Flux> sseError(SseEventPayload payload) { - String json; + public Flux> sseError(SseEventPayload errorPayload) { + String errorJson; try { - json = jsonWriter.writeValueAsString(payload); + errorJson = jsonWriter.writeValueAsString(errorPayload); } catch (JsonProcessingException serializationFailure) { log.error("Failed to serialize SSE error payload", serializationFailure); - json = ERROR_FALLBACK_JSON; + errorJson = ERROR_FALLBACK_JSON; } - ServerSentEvent errorEvent = - ServerSentEvent.builder().event(EVENT_ERROR).data(json).build(); - return Flux.deferContextual(streamContext -> { + ServerSentEvent errorEvent = ServerSentEvent.builder() + .event(EVENT_ERROR) + .data(errorJson) + .build(); + return Flux.deferContextual(streamContext -> Flux.just(errorEvent).doOnNext(emittedError -> { streamContext .getOrEmpty(StreamRequestObservation.class) .ifPresent(observation -> { @@ -278,9 +280,9 @@ public Flux> sseError(SseEventPayload payload) { .addKeyValue("event", "search_stream_failed") .addKeyValue("requestId", observation.requestId()) .addKeyValue("path", observation.path()) - .addKeyValue("code", payload.code()) - .addKeyValue("stage", payload.stage()) - .addKeyValue("retryable", payload.retryable()) + .addKeyValue("code", errorPayload.code()) + .addKeyValue("stage", errorPayload.stage()) + .addKeyValue("retryable", errorPayload.retryable()) .addKeyValue( "terminalReason", streamContext @@ -325,8 +327,7 @@ public Flux> sseError(SseEventPayload payload) { .orElse(streamFailureLog) .log(); }); - return Flux.just(errorEvent); - }); + })); } /** diff --git a/src/test/java/com/williamcallahan/javachat/adapters/in/web/ApiRequestLoggingFilterTest.java b/src/test/java/com/williamcallahan/javachat/adapters/in/web/ApiRequestLoggingFilterTest.java index e3a8b72d..6b63498c 100644 --- a/src/test/java/com/williamcallahan/javachat/adapters/in/web/ApiRequestLoggingFilterTest.java +++ b/src/test/java/com/williamcallahan/javachat/adapters/in/web/ApiRequestLoggingFilterTest.java @@ -1,21 +1,28 @@ package com.williamcallahan.javachat.adapters.in.web; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; +import ch.qos.logback.classic.Level; import ch.qos.logback.classic.Logger; +import ch.qos.logback.classic.spi.ILoggingEvent; import com.williamcallahan.javachat.support.logging.ExpectedLogEvents; import jakarta.servlet.DispatcherType; import jakarta.servlet.RequestDispatcher; import jakarta.servlet.ServletException; +import jakarta.servlet.http.HttpServletResponse; import java.io.IOException; +import java.util.List; import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; import org.slf4j.LoggerFactory; import org.springframework.mock.web.MockHttpServletRequest; import org.springframework.mock.web.MockHttpServletResponse; -/** Verifies that an escaping exception is not recorded as a successful response before error dispatch. */ +/** Verifies final HTTP completion records across normal, asynchronous, and error dispatches. */ class ApiRequestLoggingFilterTest { @Test void recordsTheErrorDispatchOutcomeInsteadOfAPrematureSuccess() throws ServletException, IOException { @@ -45,4 +52,119 @@ void recordsTheErrorDispatchOutcomeInsteadOfAPrematureSuccess() throws ServletEx && Integer.valueOf(500).equals(logField.value))); } } + + @ParameterizedTest + @CsvSource({"REQUEST,false", "REQUEST,true", "ASYNC,false", "ASYNC,true"}) + void waitsForTheRenderedErrorAfterSendErrorReturns(DispatcherType dispatchType, boolean includeMessage) + throws ServletException, IOException { + ApiRequestLoggingFilter requestLogging = new ApiRequestLoggingFilter(); + MockHttpServletRequest apiRequest = new MockHttpServletRequest("GET", "/api/chat/citations"); + MockHttpServletResponse apiResponse = new MockHttpServletResponse(); + apiRequest.setAsyncSupported(true); + try (ExpectedLogEvents capturedLogs = + ExpectedLogEvents.capture((Logger) LoggerFactory.getLogger(ApiRequestLoggingFilter.class))) { + if (dispatchType == DispatcherType.ASYNC) { + requestLogging.doFilter( + apiRequest, + apiResponse, + (servletRequest, servletResponse) -> + servletRequest.startAsync(servletRequest, servletResponse)); + assertTrue(completedRequests(capturedLogs).isEmpty()); + apiRequest.setAsyncStarted(false); + apiRequest.setDispatcherType(DispatcherType.ASYNC); + } + requestLogging.doFilter(apiRequest, apiResponse, (servletRequest, servletResponse) -> { + HttpServletResponse errorResponse = (HttpServletResponse) servletResponse; + if (includeMessage) { + errorResponse.sendError(503, "Retrieval failed"); + } else { + errorResponse.sendError(503); + } + }); + assertEquals(503, apiResponse.getStatus()); + assertTrue(completedRequests(capturedLogs).isEmpty()); + + // MockMvc does not perform the container's error-page dispatch or reopen its response. + apiResponse.setCommitted(false); + apiRequest.setDispatcherType(DispatcherType.ERROR); + apiRequest.setAttribute(RequestDispatcher.ERROR_REQUEST_URI, "/api/chat/citations"); + apiRequest.setRequestURI("/error"); + requestLogging.doFilter(apiRequest, apiResponse, (servletRequest, servletResponse) -> { + // The error page, not the status passed to sendError, owns the final response. + ((HttpServletResponse) servletResponse).setStatus(502); + ((HttpServletResponse) servletResponse).setHeader("Retry-After", "12"); + }); + assertEquals(502, apiResponse.getStatus()); + assertEquals(1, completedRequests(capturedLogs).size()); + ILoggingEvent completion = completedRequests(capturedLogs).getFirst(); + assertEquals(Level.WARN, completion.getLevel()); + assertTrue(completion.getKeyValuePairs().stream() + .anyMatch(logField -> "status".equals(logField.key) + && Integer.valueOf(502).equals(logField.value))); + assertTrue(completion.getKeyValuePairs().stream() + .anyMatch(logField -> "path".equals(logField.key) && "/api/chat/citations".equals(logField.value))); + assertTrue(completion.getKeyValuePairs().stream() + .anyMatch(logField -> "requestId".equals(logField.key) + && apiResponse.getHeader("X-Request-ID").equals(logField.value))); + assertTrue(completion.getKeyValuePairs().stream() + .anyMatch(logField -> "retryAfter".equals(logField.key) && "12".equals(logField.value))); + assertTrue(completion.getKeyValuePairs().stream() + .anyMatch(logField -> "durationMs".equals(logField.key) + && logField.value instanceof Long durationMillis + && durationMillis >= 0)); + } + } + + @Test + void recordsANestedErrorDispatchOnlyOnce() throws ServletException, IOException { + ApiRequestLoggingFilter requestLogging = new ApiRequestLoggingFilter(); + MockHttpServletRequest apiRequest = new MockHttpServletRequest("GET", "/api/chat/citations"); + MockHttpServletResponse apiResponse = new MockHttpServletResponse(); + try (ExpectedLogEvents capturedLogs = + ExpectedLogEvents.capture((Logger) LoggerFactory.getLogger(ApiRequestLoggingFilter.class))) { + requestLogging.doFilter(apiRequest, apiResponse, (servletRequest, servletResponse) -> { + ((HttpServletResponse) servletResponse).sendError(503); + apiResponse.setCommitted(false); + apiRequest.setDispatcherType(DispatcherType.ERROR); + apiRequest.setAttribute(RequestDispatcher.ERROR_REQUEST_URI, "/api/chat/citations"); + apiRequest.setRequestURI("/error"); + requestLogging.doFilter( + apiRequest, apiResponse, (errorRequest, errorResponse) -> ((HttpServletResponse) errorResponse) + .setStatus(502)); + assertTrue(completedRequests(capturedLogs).isEmpty()); + apiRequest.setDispatcherType(DispatcherType.REQUEST); + ((HttpServletResponse) servletResponse).setStatus(500); + }); + assertEquals(1, completedRequests(capturedLogs).size()); + assertEquals(500, apiResponse.getStatus()); + assertTrue(completedRequests(capturedLogs).getFirst().getKeyValuePairs().stream() + .anyMatch(logField -> "status".equals(logField.key) + && Integer.valueOf(500).equals(logField.value))); + } + } + + @Test + void recordsARegularResponseOnceWithoutAnErrorDispatch() throws ServletException, IOException { + ApiRequestLoggingFilter requestLogging = new ApiRequestLoggingFilter(); + MockHttpServletRequest apiRequest = new MockHttpServletRequest("GET", "/api/chat/citations"); + MockHttpServletResponse apiResponse = new MockHttpServletResponse(); + try (ExpectedLogEvents capturedLogs = + ExpectedLogEvents.capture((Logger) LoggerFactory.getLogger(ApiRequestLoggingFilter.class))) { + requestLogging.doFilter(apiRequest, apiResponse, (servletRequest, servletResponse) -> ((HttpServletResponse) + servletResponse) + .setStatus(204)); + assertFalse(apiRequest.isAsyncStarted()); + assertEquals(1, completedRequests(capturedLogs).size()); + assertEquals(Level.INFO, completedRequests(capturedLogs).getFirst().getLevel()); + assertTrue(completedRequests(capturedLogs).getFirst().getKeyValuePairs().stream() + .anyMatch(logField -> "status".equals(logField.key) + && Integer.valueOf(204).equals(logField.value))); + } + } + + private static List completedRequests(ExpectedLogEvents capturedLogs) { + return capturedLogs.events().stream() + .filter(logEvent -> "API request completed".equals(logEvent.getFormattedMessage())) + .toList(); + } } diff --git a/src/test/java/com/williamcallahan/javachat/config/SecurityConfigTest.java b/src/test/java/com/williamcallahan/javachat/config/SecurityConfigTest.java index ab0511ea..bc31712b 100644 --- a/src/test/java/com/williamcallahan/javachat/config/SecurityConfigTest.java +++ b/src/test/java/com/williamcallahan/javachat/config/SecurityConfigTest.java @@ -23,6 +23,8 @@ import com.williamcallahan.javachat.application.auth.VerifiedApiKey; import com.williamcallahan.javachat.support.logging.ExpectedLogEvents; import com.williamcallahan.javachat.web.CsrfController; +import jakarta.servlet.DispatcherType; +import jakarta.servlet.RequestDispatcher; import jakarta.servlet.http.Cookie; import java.util.Optional; import org.junit.jupiter.api.Test; @@ -80,6 +82,9 @@ class SecurityConfigTest { @Autowired MockMvc mockMvc; + @Autowired + ApiRequestLoggingFilter requestLogging; + @MockitoBean ClerkApiKeyVerifier clerkApiKeyVerifier; @@ -185,10 +190,46 @@ void recordsStreamingCompletionAfterTheAsynchronousResponseFinishes() throws Exc @Test void returnsJsonWhenCsrfTokenIsMissing() throws Exception { - mockMvc.perform(post(CSRF_PROTECTED_ENDPOINT)) - .andExpect(status().isForbidden()) - .andExpect(content().contentTypeCompatibleWith(MediaType.APPLICATION_JSON)) - .andExpect(jsonPath("$.message").value(CSRF_INVALID_MESSAGE)); + try (ExpectedLogEvents capturedLogs = + ExpectedLogEvents.capture((Logger) LoggerFactory.getLogger(ApiRequestLoggingFilter.class))) { + mockMvc.perform(post(CSRF_PROTECTED_ENDPOINT)) + .andExpect(status().isForbidden()) + .andExpect(content().contentTypeCompatibleWith(MediaType.APPLICATION_JSON)) + .andExpect(header().exists("X-Request-ID")) + .andExpect(jsonPath("$.message").value(CSRF_INVALID_MESSAGE)); + var completedRequests = capturedLogs.events().stream() + .filter(logEvent -> "API request completed".equals(logEvent.getFormattedMessage())) + .toList(); + assertEquals(1, completedRequests.size()); + assertTrue(completedRequests.getFirst().getKeyValuePairs().stream() + .anyMatch(logField -> "status".equals(logField.key) + && Integer.valueOf(403).equals(logField.value))); + } + } + + @Test + void waitsForSecurityEntryPointErrorRendering() throws Exception { + try (ExpectedLogEvents capturedLogs = + ExpectedLogEvents.capture((Logger) LoggerFactory.getLogger(ApiRequestLoggingFilter.class))) { + MvcResult deniedExchange = mockMvc.perform(get(API_KEY_SUBJECT_ENDPOINT)) + .andExpect(status().isForbidden()) + .andExpect(header().exists("X-Request-ID")) + .andReturn(); + assertTrue(capturedLogs.events().stream() + .noneMatch(logEvent -> "API request completed".equals(logEvent.getFormattedMessage()))); + deniedExchange.getRequest().setDispatcherType(DispatcherType.ERROR); + deniedExchange.getRequest().setAttribute(RequestDispatcher.ERROR_REQUEST_URI, API_KEY_SUBJECT_ENDPOINT); + deniedExchange.getRequest().setRequestURI("/error"); + requestLogging.doFilter( + deniedExchange.getRequest(), deniedExchange.getResponse(), (errorRequest, errorResponse) -> {}); + var completedRequests = capturedLogs.events().stream() + .filter(logEvent -> "API request completed".equals(logEvent.getFormattedMessage())) + .toList(); + assertEquals(1, completedRequests.size()); + assertTrue(completedRequests.getFirst().getKeyValuePairs().stream() + .anyMatch(logField -> "status".equals(logField.key) + && Integer.valueOf(403).equals(logField.value))); + } } @Test diff --git a/src/test/java/com/williamcallahan/javachat/web/SseSupportTest.java b/src/test/java/com/williamcallahan/javachat/web/SseSupportTest.java index 3ebcea75..ce3a4899 100644 --- a/src/test/java/com/williamcallahan/javachat/web/SseSupportTest.java +++ b/src/test/java/com/williamcallahan/javachat/web/SseSupportTest.java @@ -438,4 +438,34 @@ void citationPartialFailureStatusFluxEmitsNothingWhenAllCitationsConvert() { private SseSupport createSseSupport() { return new SseSupport(objectMapper, new SimpleMeterRegistry()); } + + @Test + void terminalErrorDiagnosticsWaitForDemandAndRemainAbsentAfterCancellation() { + SseSupport sseSupport = createSseSupport(); + MockHttpServletResponse servletResponse = new MockHttpServletResponse(); + servletResponse.setHeader("X-Request-ID", "demand-correlated-request"); + Flux> failedStream = sseSupport.withRequestDiagnostics( + sseSupport.configuredProviderUnavailableError(), servletResponse, "/api/chat/stream"); + try (ExpectedLogEvents capturedLogs = + ExpectedLogEvents.capture((Logger) LoggerFactory.getLogger(SseSupport.class))) { + StepVerifier.create(failedStream, 0) + .then(() -> assertTrue(capturedLogs.events().isEmpty())) + .thenCancel() + .verify(); + assertTrue(capturedLogs.events().isEmpty()); + + StepVerifier.create(failedStream, 0) + .then(() -> assertTrue(capturedLogs.events().isEmpty())) + .thenRequest(1) + .assertNext(errorEvent -> { + assertEquals(EVENT_ERROR, errorEvent.event()); + assertEquals(1, capturedLogs.events().size()); + assertTrue(capturedLogs.events().getFirst().getKeyValuePairs().stream() + .anyMatch(logField -> "requestId".equals(logField.key) + && "demand-correlated-request".equals(logField.value))); + }) + .verifyComplete(); + assertEquals(1, capturedLogs.events().size()); + } + } } From 40220034f0a2e6a2b04719e7db4d152a93d04d12 Mon Sep 17 00:00:00 2001 From: William Callahan Date: Sun, 1 Nov 2026 22:30:45 -0800 Subject: [PATCH 8/9] docs(diagnostics): explain deferred error response completion Document why the servlet response boundary tracks pending error rendering. Agent-Session: ses_e651faff3ffeDdTophrwg3Q2V7 --- .../javachat/adapters/in/web/ApiRequestLoggingFilter.java | 1 + 1 file changed, 1 insertion(+) diff --git a/src/main/java/com/williamcallahan/javachat/adapters/in/web/ApiRequestLoggingFilter.java b/src/main/java/com/williamcallahan/javachat/adapters/in/web/ApiRequestLoggingFilter.java index 220b504b..e5849855 100644 --- a/src/main/java/com/williamcallahan/javachat/adapters/in/web/ApiRequestLoggingFilter.java +++ b/src/main/java/com/williamcallahan/javachat/adapters/in/web/ApiRequestLoggingFilter.java @@ -106,6 +106,7 @@ protected void doFilterNestedErrorDispatch( apiRequest.removeAttribute(ERROR_PENDING_ATTRIBUTE); } + /** Distinguishes container-rendered errors from responses that finish within the current dispatch. */ private static final class ErrorDispatchResponse extends HttpServletResponseWrapper { private final HttpServletRequest apiRequest; From bee5ced46d558fd4049d81e7bb45f30fc5de3fb0 Mon Sep 17 00:00:00 2001 From: William Callahan Date: Sun, 1 Nov 2026 22:35:35 -0800 Subject: [PATCH 9/9] test(diagnostics): require the correlated request header Make the non-null header expectation explicit before matching its log field. Agent-Session: ses_e651faff3ffeDdTophrwg3Q2V7 --- .../adapters/in/web/ApiRequestLoggingFilterTest.java | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/src/test/java/com/williamcallahan/javachat/adapters/in/web/ApiRequestLoggingFilterTest.java b/src/test/java/com/williamcallahan/javachat/adapters/in/web/ApiRequestLoggingFilterTest.java index 6b63498c..cd0d7ccf 100644 --- a/src/test/java/com/williamcallahan/javachat/adapters/in/web/ApiRequestLoggingFilterTest.java +++ b/src/test/java/com/williamcallahan/javachat/adapters/in/web/ApiRequestLoggingFilterTest.java @@ -15,6 +15,7 @@ import jakarta.servlet.http.HttpServletResponse; import java.io.IOException; import java.util.List; +import java.util.Objects; import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.CsvSource; @@ -97,6 +98,7 @@ void waitsForTheRenderedErrorAfterSendErrorReturns(DispatcherType dispatchType, assertEquals(502, apiResponse.getStatus()); assertEquals(1, completedRequests(capturedLogs).size()); ILoggingEvent completion = completedRequests(capturedLogs).getFirst(); + String requestId = Objects.requireNonNull(apiResponse.getHeader("X-Request-ID")); assertEquals(Level.WARN, completion.getLevel()); assertTrue(completion.getKeyValuePairs().stream() .anyMatch(logField -> "status".equals(logField.key) @@ -104,8 +106,7 @@ void waitsForTheRenderedErrorAfterSendErrorReturns(DispatcherType dispatchType, assertTrue(completion.getKeyValuePairs().stream() .anyMatch(logField -> "path".equals(logField.key) && "/api/chat/citations".equals(logField.value))); assertTrue(completion.getKeyValuePairs().stream() - .anyMatch(logField -> "requestId".equals(logField.key) - && apiResponse.getHeader("X-Request-ID").equals(logField.value))); + .anyMatch(logField -> "requestId".equals(logField.key) && requestId.equals(logField.value))); assertTrue(completion.getKeyValuePairs().stream() .anyMatch(logField -> "retryAfter".equals(logField.key) && "12".equals(logField.value))); assertTrue(completion.getKeyValuePairs().stream()