diff --git a/docs/api.md b/docs/api.md index 19a527ec..85ed02b6 100644 --- a/docs/api.md +++ b/docs/api.md @@ -50,6 +50,26 @@ 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. +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 + +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..047c3895 --- /dev/null +++ b/frontend/e2e/search-diagnostics.spec.ts @@ -0,0 +1,47 @@ +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 + .locator(".message.assistant.error .bubble") + .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..4c8d35a0 --- /dev/null +++ b/frontend/src/lib/services/searchDiagnostics.test.ts @@ -0,0 +1,174 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import { BrowserSearchFailureSchema } from "../validation/schemas"; +import { streamSse, streamSseGet } from "./sse"; + +describe("search failure diagnostic transport", () => { + afterEach(() => { + window.history.replaceState({}, "", "/"); + vi.unstubAllGlobals(); + vi.restoreAllMocks(); + }); + + 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( + "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"); + expect(submittedReports[0]).not.toContain("private-query"); + expect(submittedReports[0]).not.toContain("private-fragment"); + 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, + 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( + "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..446708da 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"; @@ -27,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 { @@ -43,6 +46,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 +261,7 @@ export async function streamSse( requestOptions: StreamSseRequestOptions = {}, ): Promise { const abortSignal = requestOptions.signal; + const startedAt = performance.now(); let httpResponse: Response; try { @@ -275,10 +282,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 +304,7 @@ export async function streamSseGet( requestOptions: StreamSseRequestOptions = {}, ): Promise { const abortSignal = requestOptions.signal; + const startedAt = performance.now(); let httpResponse: Response; try { @@ -299,10 +313,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 && httpResponse.body !== null) { + 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.slice(0, DIAGNOSTIC_PAGE_PATH_MAX_LENGTH), + frontendAsset: frontendScript ? new URL(frontendScript.src).pathname : undefined, + elapsedMs: Math.round(performance.now() - startedAt), + }); } async function consumeSseStream( @@ -312,17 +365,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..13dbb2dd 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.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..e5849855 --- /dev/null +++ b/src/main/java/com/williamcallahan/javachat/adapters/in/web/ApiRequestLoggingFilter.java @@ -0,0 +1,137 @@ +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; +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_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; + + /** Excludes static assets and actuator scrapes from application request records. */ + @Override + protected boolean shouldNotFilter(HttpServletRequest apiRequest) { + 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. */ + @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(requestPath(apiRequest), 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.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(); + } + 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() + && (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") + .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(); + } + } + + /** 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); + } + + /** Distinguishes container-rendered errors from responses that finish within the current dispatch. */ + 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/adapters/in/web/BrowserSearchFailureController.java b/src/main/java/com/williamcallahan/javachat/adapters/in/web/BrowserSearchFailureController.java new file mode 100644 index 00000000..bc288f48 --- /dev/null +++ b/src/main/java/com/williamcallahan/javachat/adapters/in/web/BrowserSearchFailureController.java @@ -0,0 +1,93 @@ +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", + StructuredLogValue.bounded(failureReport.path(), MAX_DIAGNOSTIC_FIELD_LENGTH) + .text()) + .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", + 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(); + } + + /** 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..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 @@ -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,9 @@ enum AlertField { UPSTREAM_HTTP_STATUS("upstreamHttpStatus"), UPSTREAM_CODE("upstreamCode"), UPSTREAM_TYPE("upstreamType"), + GATEWAY_REQUEST_ID("gatewayRequestId"), + RETRY_AFTER("retryAfter"), + TERMINAL_REASON("terminalReason"), QUEUE_TIER("queueTier"), QUEUE_LANE("queueLane"), QUEUE_TIMEOUT_PHASE("queueTimeoutPhase"), @@ -112,6 +116,9 @@ String wireName() { private final Integer upstreamHttpStatus; private final String upstreamCode; 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) { @@ -124,15 +131,25 @@ 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); 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)); @@ -159,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; @@ -251,6 +274,15 @@ 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); + } + 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/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..b81f477b 100644 --- a/src/main/java/com/williamcallahan/javachat/web/ChatController.java +++ b/src/main/java/com/williamcallahan/javachat/web/ChatController.java @@ -228,55 +228,63 @@ 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.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 4b9b174b..8506eedc 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"); } /** @@ -417,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 c16921ec..7ca219e4 100644 --- a/src/main/java/com/williamcallahan/javachat/web/SseSupport.java +++ b/src/main/java/com/williamcallahan/javachat/web/SseSupport.java @@ -6,7 +6,10 @@ 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.application.streaming.ReportedStreamingFailure; 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 +24,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 +43,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 = @@ -125,6 +130,36 @@ 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) {} + + /** 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 correlatedErrors.contextWrite( + streamContext -> streamContext.put(OpenAIServiceException.class, providerFailure)); + } + failureCause = failureCause.getCause(); + } + return correlatedErrors; + } + /** * Prepares a raw content stream for SSE transmission with backpressure handling. * Filters empty chunks and applies the standard streaming configuration. @@ -224,16 +259,75 @@ 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; } - return Flux.just( - ServerSentEvent.builder().event(EVENT_ERROR).data(json).build()); + 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 -> { + 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", errorPayload.code()) + .addKeyValue("stage", errorPayload.stage()) + .addKeyValue("retryable", errorPayload.retryable()) + .addKeyValue( + "terminalReason", + streamContext + .getOrEmpty(ReportedStreamingFailure.class) + .flatMap(ReportedStreamingFailure::terminalReason) + .orElse(null)) + .addKeyValue( + "durationMs", + java.util.concurrent.TimeUnit.NANOSECONDS.toMillis( + System.nanoTime() - observation.startedNanos())); + streamContext + .getOrEmpty(OpenAIServiceException.class) + .map(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())) + .orElse(streamFailureLog) + .log(); + }); + })); } /** 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..cd0d7ccf --- /dev/null +++ b/src/test/java/com/williamcallahan/javachat/adapters/in/web/ApiRequestLoggingFilterTest.java @@ -0,0 +1,171 @@ +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 java.util.Objects; +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 final HTTP completion records across normal, asynchronous, and error dispatches. */ +class ApiRequestLoggingFilterTest { + @Test + void recordsTheErrorDispatchOutcomeInsteadOfAPrematureSuccess() 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))) { + 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))); + } + } + + @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(); + String requestId = Objects.requireNonNull(apiResponse.getHeader("X-Request-ID")); + 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) && requestId.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/adapters/in/web/BrowserSearchFailureControllerTest.java b/src/test/java/com/williamcallahan/javachat/adapters/in/web/BrowserSearchFailureControllerTest.java new file mode 100644 index 00000000..47781c9e --- /dev/null +++ b/src/test/java/com/williamcallahan/javachat/adapters/in/web/BrowserSearchFailureControllerTest.java @@ -0,0 +1,117 @@ +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; +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; + + @Autowired + ObjectMapper objectMapper; + + @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()); + } + + @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/adapters/out/llm/openai/OpenAiStreamingFailureExceptionTest.java b/src/test/java/com/williamcallahan/javachat/adapters/out/llm/openai/OpenAiStreamingFailureExceptionTest.java index e42615a3..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 @@ -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"); @@ -187,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/config/SecurityConfigTest.java b/src/test/java/com/williamcallahan/javachat/config/SecurityConfigTest.java index 569c3116..bc31712b 100644 --- a/src/test/java/com/williamcallahan/javachat/config/SecurityConfigTest.java +++ b/src/test/java/com/williamcallahan/javachat/config/SecurityConfigTest.java @@ -6,21 +6,29 @@ 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.DispatcherType; +import jakarta.servlet.RequestDispatcher; 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 +39,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. @@ -72,6 +82,9 @@ class SecurityConfigTest { @Autowired MockMvc mockMvc; + @Autowired + ApiRequestLoggingFilter requestLogging; + @MockitoBean ClerkApiKeyVerifier clerkApiKeyVerifier; @@ -134,12 +147,89 @@ 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)) - .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 @@ -252,6 +342,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/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/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/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 2bf966ea..ce3a4899 100644 --- a/src/test/java/com/williamcallahan/javachat/web/SseSupportTest.java +++ b/src/test/java/com/williamcallahan/javachat/web/SseSupportTest.java @@ -12,23 +12,34 @@ 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.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; import io.micrometer.core.instrument.simple.SimpleMeterRegistry; 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; 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 +59,115 @@ 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 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))); + } + } + } + + @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(); + 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(); @@ -318,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()); + } + } }