From 3600b0474ab9b7cff0dffce265022c74942a9596 Mon Sep 17 00:00:00 2001 From: diegosouzapw <8016841+diegosouzapw@users.noreply.github.com> Date: Fri, 24 Jul 2026 00:37:44 -0300 Subject: [PATCH] refactor(sse): classify SSE critical-path empty catches + add CONTRIBUTING convention (#8142) --- CONTRIBUTING.md | 25 ++++ .../features/8142-empty-catch-convention.md | 1 + open-sse/utils/stream.ts | 18 +-- open-sse/utils/streamHandler.ts | 9 +- .../stream-onfailure-callback-logging.test.ts | 119 ++++++++++++++++++ 5 files changed, 161 insertions(+), 11 deletions(-) create mode 100644 changelog.d/features/8142-empty-catch-convention.md create mode 100644 tests/unit/stream-onfailure-callback-logging.test.ts diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index d5395a97ccdc..f2cd27ac57b9 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -228,6 +228,31 @@ Current test status: **122 unit test files** covering: - **Zod validation** — Use Zod v4 schemas for all API input validation - **Naming**: Files = camelCase/kebab-case, components = PascalCase, constants = UPPER_SNAKE +### Error handling / empty catch blocks + +Never leave a `catch` unexplained. Classify it into one of two buckets (operationalizes +the hard rule "never silently swallow errors in SSE streams"): + +- **Intentional (our own best-effort cleanup/telemetry)** — a failure here is expected and + harmless; add a one-line rationale comment, no logging (logging on every request is the + noise this convention avoids). + + ```ts + } catch {} // closing an already-closed controller after client disconnect is expected + ``` + +- **Should log (external/caller-supplied code, or the swallow changes control flow)** — keep + the catch (never let it break the stream) but emit a contextual `console.debug`/`warn` so the + failure is discoverable. + + ```ts + } catch (e) { + console.debug("[STREAM] onFailure callback error:", e); + } + ``` + +See `open-sse/utils/stream.ts` and `open-sse/utils/streamHandler.ts` for applied examples. + --- ## Project Structure diff --git a/changelog.d/features/8142-empty-catch-convention.md b/changelog.d/features/8142-empty-catch-convention.md new file mode 100644 index 000000000000..5d38572dc38b --- /dev/null +++ b/changelog.d/features/8142-empty-catch-convention.md @@ -0,0 +1 @@ +- refactor(sse): classify SSE critical-path empty catches + add CONTRIBUTING convention (#8142) diff --git a/open-sse/utils/stream.ts b/open-sse/utils/stream.ts index c28fde029d9a..60fe7d7fe95e 100644 --- a/open-sse/utils/stream.ts +++ b/open-sse/utils/stream.ts @@ -641,9 +641,7 @@ function restoreClaudePassthroughToolUseName(parsed: JsonRecord, toolNameMap: un // to avoid shared state issues with concurrent streams (TextDecoder with {stream:true} // maintains internal buffering state between decode() calls). -/** - * Stream modes - */ +// Stream modes const STREAM_MODE = { TRANSLATE: "translate", // Full translation between formats PASSTHROUGH: "passthrough", // No translation, normalize output, extract usage @@ -813,7 +811,7 @@ export function createSSEStream(options: StreamOptions = {}) { let pendingToolFinishTime: number | null = null; try { pendingToolFinishTime = consumeToolFinishTime(sessionId); - } catch {} + } catch {} // best-effort read of optional timing state — absence is normal // Guard against duplicate [DONE] events — ensures exactly one per stream let doneSent = false; @@ -1830,7 +1828,7 @@ export function createSSEStream(options: StreamOptions = {}) { toolTs ? now - toolTs : null, lastChunkTs ? now - lastChunkTs : null ); - } catch {} + } catch {} // best-effort telemetry — must never break the stream pendingToolFinishTime = null; } } @@ -1867,7 +1865,7 @@ export function createSSEStream(options: StreamOptions = {}) { toolFinishTime = now; try { markToolFinish(sessionId); - } catch {} + } catch {} // best-effort bookkeeping write — a miss just skips latency correlation } // T18: Normalize finish_reason to 'tool_calls' if tool calls were used @@ -1943,7 +1941,9 @@ export function createSSEStream(options: StreamOptions = {}) { if (onFailure) { try { failureHandled = onFailure(failurePayload) === true; - } catch {} + } catch (e) { + console.debug(`[STREAM] onFailure callback error:`, e); + } } clearIdleTimer(); if (!failureHandled) { @@ -1995,7 +1995,7 @@ export function createSSEStream(options: StreamOptions = {}) { toolFinishTime = now; try { markToolFinish(sessionId); - } catch {} + } catch {} // best-effort bookkeeping write — a miss just skips latency correlation } // Track content length and accumulate for call log (from raw provider chunk, so content is never missed) @@ -2109,7 +2109,7 @@ export function createSSEStream(options: StreamOptions = {}) { toolTs ? now - toolTs : null, lastChunkTs ? now - lastChunkTs : null ); - } catch {} + } catch {} // best-effort telemetry — must never break the stream pendingToolFinishTime = null; } } diff --git a/open-sse/utils/streamHandler.ts b/open-sse/utils/streamHandler.ts index d4b4df86b8c2..e3f4ebaa1c28 100644 --- a/open-sse/utils/streamHandler.ts +++ b/open-sse/utils/streamHandler.ts @@ -247,7 +247,10 @@ export function createStreamController({ try { trackPendingRequest(model || "", provider || "", connectionId ?? null, false); } catch (e) { - console.error(`[${getTimeString()}] [streamHandler] trackPendingRequest decrement failed — counter may drift`, e); + console.error( + `[${getTimeString()}] [streamHandler] trackPendingRequest decrement failed — counter may drift`, + e + ); } }; @@ -583,7 +586,9 @@ export function createDisconnectAwareStream(transformStream, streamController) { try { controller.close(); - } catch {} + } catch { + // Closing an already-closed/aborted controller after client disconnect is expected. + } } }, diff --git a/tests/unit/stream-onfailure-callback-logging.test.ts b/tests/unit/stream-onfailure-callback-logging.test.ts new file mode 100644 index 000000000000..8e9a0f767279 --- /dev/null +++ b/tests/unit/stream-onfailure-callback-logging.test.ts @@ -0,0 +1,119 @@ +import test from "node:test"; +import assert from "node:assert/strict"; +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; + +// #8142 — the empty `catch {}` around the caller-supplied `onFailure(failurePayload)` +// callback in createSSEStream (open-sse/utils/stream.ts) silently swallowed a throw +// from consumer code, leaving `failureHandled` false with zero diagnostic trace. +// Fix: keep the catch (a buggy failure handler must never break the stream) but emit +// a contextual console.debug so the swallowed error is discoverable. + +const TEST_DATA_DIR = fs.mkdtempSync(path.join(os.tmpdir(), "omniroute-8142-")); +process.env.DATA_DIR = TEST_DATA_DIR; +const core = await import("../../src/lib/db/core.ts"); +const { createSSEStream } = await import("../../open-sse/utils/stream.ts"); +const { FORMATS } = await import("../../open-sse/translator/formats.ts"); + +const enc = new TextEncoder(); + +async function readTransformed(chunks: string[], options: Record) { + const source = new ReadableStream({ + start(c) { + for (const chunk of chunks) c.enqueue(enc.encode(chunk)); + c.close(); + }, + }); + return new Response( + source.pipeThrough(createSSEStream(options as Parameters[0])) + ).text(); +} + +test.after(() => { + core.resetDbInstance(); + if (fs.existsSync(TEST_DATA_DIR)) { + for (const entry of fs.readdirSync(TEST_DATA_DIR)) { + fs.rmSync(path.join(TEST_DATA_DIR, entry), { recursive: true, force: true }); + } + } +}); + +// An OpenAI-Responses-style `response.failed` passthrough event reliably drives +// createSSEStream into the `failurePayload` branch (~L1941-1953) that invokes the +// caller-supplied `onFailure` callback — reused here as the trigger. +function responseFailedChunk(message: string) { + return `data: ${JSON.stringify({ + type: "response.failed", + response: { error: { message, code: "server_error" } }, + })}\n\n`; +} + +test("#8142 onFailure throwing does not crash the stream and is logged", async () => { + const originalDebug = console.debug; + const debugCalls: unknown[][] = []; + console.debug = (...args: unknown[]) => { + debugCalls.push(args); + }; + try { + await assert.rejects( + readTransformed([responseFailedChunk("boom upstream")], { + mode: "passthrough", + sourceFormat: FORMATS.OPENAI, + provider: "openai", + model: "gpt-test", + body: { messages: [{ role: "user", content: "hello" }] }, + onFailure() { + throw new Error("boom from consumer onFailure handler"); + }, + }), + /boom upstream/i, + "stream must still reject normally — the callback throw must not corrupt control flow" + ); + } finally { + console.debug = originalDebug; + } + const loggedOnFailureThrow = debugCalls.some((args) => + args.some((a) => typeof a === "string" && /onFailure/i.test(a)) + ); + assert.ok( + loggedOnFailureThrow, + "a console.debug call referencing onFailure must be emitted when the callback throws" + ); +}); + +test("#8142 regression: onFailure returning normally logs nothing and behaves identically", async () => { + const originalDebug = console.debug; + const debugCalls: unknown[][] = []; + console.debug = (...args: unknown[]) => { + debugCalls.push(args); + }; + let failurePayload: Record | null = null; + try { + await assert.rejects( + readTransformed([responseFailedChunk("boom upstream normal")], { + mode: "passthrough", + sourceFormat: FORMATS.OPENAI, + provider: "openai", + model: "gpt-test", + body: { messages: [{ role: "user", content: "hello" }] }, + onFailure(p: Record) { + failurePayload = p; + return true; + }, + }), + /boom upstream normal/i + ); + } finally { + console.debug = originalDebug; + } + assert.ok(failurePayload, "onFailure callback must still be invoked normally"); + const loggedOnFailureThrow = debugCalls.some((args) => + args.some((a) => typeof a === "string" && /onFailure/i.test(a)) + ); + assert.equal( + loggedOnFailureThrow, + false, + "the happy path (no throw) must not emit the onFailure-throw debug log — behavior-free regression guard" + ); +});