fix(proxy): omit terminal [DONE] when stream terminates abnormally - #198
Merged
Conversation
|
Warning Rate limit exceeded
You’ve run out of usage credits. Purchase more in the billing tab. ⌛ How to resolve this issue?After the wait time has elapsed, a review can be triggered using the We recommend that you space out your commits to avoid hitting the rate limit. 🚦 How do rate limits work?CodeRabbit enforces hourly rate limits for each developer per organization. Our paid plans have higher rate limits than the trial, open-source and free plans. In all cases, we re-allow further reviews after a brief timeout. Please see our FAQ for further information. ℹ️ Review info⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Free Run ID: 📒 Files selected for processing (3)
📝 WalkthroughWalkthroughWhen an upstream streaming connection errors or disconnects mid-stream, the SSE builder sets an ChangesStreaming Error Handling
🎯 3 (Moderate) | ⏱️ ~20 minutes Note 🎁 Summarized by CodeRabbit FreeYour organization is on the Free plan. CodeRabbit will generate a high-level summary and a walkthrough for each pull request. For a comprehensive line-by-line review, please upgrade your subscription to CodeRabbit Pro by visiting https://app.coderabbit.ai/login. Comment |
There was a problem hiding this comment.
Pull request overview
Fixes the proxy’s SSE streaming termination semantics to match docs/api-proxy.md §5: when an upstream stream errors mid-flight, the proxy must emit an error chunk and close without the terminal data: [DONE], so client SDKs don’t misinterpret truncated streams as clean completions.
Changes:
- Track whether an SSE
event: errorwas emitted during streaming; only append terminal[DONE]on clean completion. - Add an E2E regression test covering “upstream disconnects mid-stream” to assert partial chunk forwarding + iterator-time error + no
[DONE]/ no syntheticfinish_reason:"stop".
Reviewed changes
Copilot reviewed 2 out of 2 changed files in this pull request and generated 1 comment.
| File | Description |
|---|---|
crates/aisix-proxy/src/chat.rs |
Conditionally omit terminal [DONE] when a streaming error was yielded. |
tests/e2e/src/cases/streaming-edges-e2e.test.ts |
Adds an E2E case for upstream mid-stream disconnects asserting the documented streaming contract. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Comment on lines
184
to
+188
| }); | ||
|
|
||
| test("upstream disconnects mid-stream: partial chunks reach caller, iterator throws, no [DONE]", async (ctx) => { | ||
| if (!etcdReachable || !app || !admin) { | ||
| ctx.skip(); |
moonming
added a commit
that referenced
this pull request
May 10, 2026
…unit test Audit follow-ups on PR #198: - MEDIUM-1: SSE `event: error` frames previously carried a plain string `data:` payload. The OpenAI Node SDK calls `JSON.parse(sse.data)` BEFORE checking `sse.event === "error"`, so a plain-string payload yields a SyntaxError instead of the typed APIError callers expect. Emit the OpenAI error-envelope shape `{"error": {"message": "...", "type": "..."}}` per https://platform.openai.com/docs/guides/error-codes/api-errors. The error_type field is sourced from the existing ProxyError::error_type() taxonomy for upstream errors and defaults to "internal_error" for fatal-stream-encode errors. - HIGH-2: add Rust unit test streaming_response_omits_done_when_upstream_returns_invalid_json_mid_stream that drains raw response bytes and asserts: * !contains("data: [DONE]") — no terminal sentinel after error * contains("event: error") — error event emitted * data line parses as JSON with `error` key — envelope shape This pins the contract at the byte level, independent of any SDK parser behavior. - MEDIUM-2: bump streaming-edges-e2e disconnect case `eventDelayMs` 50→200 to match the peer client-abort case. Verification: - cargo test -p aisix-proxy --lib: 139/139 passing (incl. new test) - cargo clippy -p aisix-proxy --tests -- -D warnings: clean - cargo fmt --check: clean
moonming
added a commit
that referenced
this pull request
May 10, 2026
…d APIError from SyntaxError Audit MEDIUM-2 follow-up on PR #198: The prior `surfacedError === true` assertion was satisfied by EITHER: - The new typed `APIError` path — OpenAI Node SDK throws `APIError` after parsing the SSE error frame's JSON envelope - The OLD plain-string path — SDK calls `JSON.parse(sse.data)` BEFORE checking `sse.event === "error"`, so a plain-string payload yields `SyntaxError("Could not parse message into JSON: ...")` A regression that re-introduced `err.to_string()` raw on the SSE frame would still pass the prior assertion. Tighten by capturing the thrown error's class and message and asserting: - `errCtor !== "SyntaxError"` — must not be the SDK parser-error path - `!errMessage.includes("Could not parse message into JSON")` — must not be the SDK parser-error message This pins the user-visible contract that callers see a typed `APIError`, not a spurious parser failure, on abnormal stream termination.
…loses #177) Issue: per docs/api-proxy.md §5 ("Streaming protocol details"): > "If the upstream stream terminates abnormally, aisix sends a > final error chunk and closes the response without `[DONE]`." The proxy's `build_sse_stream` (crates/aisix-proxy/src/chat.rs) correctly emitted an SSE `event: error` chunk on upstream errors but then unconditionally appended `data: [DONE]\n\n` afterward — making the truncated stream look complete to SDK consumers that treat `[DONE]` as the clean-completion signal. Per the same docs §5: "Client SDKs that interpret missing `[DONE]` as an error will surface the right error class" — that signal was being silently overwritten. Fix: track whether an error event has been yielded during the stream loop; only emit the terminal `[DONE]` on a clean upstream completion. The `on_complete` token-accounting callback still fires unconditionally so TPM/usage telemetry isn't lost on the abnormal-termination path. E2E regression test (tests/e2e/src/cases/streaming-edges-e2e.test.ts): restored "upstream disconnects mid-stream" case from the held-back queue. Mock upstream emits 2 SSE chunks then `res.destroy()`s the TCP connection (no `[DONE]`). Strict assertions per docs §5: - collected.join("") === "partial " (chunks 1 + 2 reach caller) - sawFinish === false (no synthetic `finish_reason: "stop"` injection — chunk 4 with the real stop never reached gateway) - surfacedError === true (iterator throws on the SSE error event, since OpenAI Node SDK treats `event: error` as a fatal stream error) `eventDelayMs: 50` between mock writes ensures both chunks flush to the gateway BEFORE `res.destroy()` aborts the connection — without the delay, write buffering on the mock side races the destroy and the gateway sometimes sees zero chunks (which would surface as a request-time 502 instead of an iterator error, masking the regression test's intent). Verified locally: - cargo test -p aisix-proxy --lib: 138/138 - cargo clippy --workspace -- -D warnings: clean - streaming-edges-e2e.test.ts: 2/2 (client abort + upstream disconnect) Note: full e2e suite has an unrelated failure in weighted-routing-distribution-e2e (filed as #197 — pre-existing bug on main where 100/0 split lands instead of 70/30). This PR does not touch routing; verified that the same test fails on pristine main.
…unit test Audit follow-ups on PR #198: - MEDIUM-1: SSE `event: error` frames previously carried a plain string `data:` payload. The OpenAI Node SDK calls `JSON.parse(sse.data)` BEFORE checking `sse.event === "error"`, so a plain-string payload yields a SyntaxError instead of the typed APIError callers expect. Emit the OpenAI error-envelope shape `{"error": {"message": "...", "type": "..."}}` per https://platform.openai.com/docs/guides/error-codes/api-errors. The error_type field is sourced from the existing ProxyError::error_type() taxonomy for upstream errors and defaults to "internal_error" for fatal-stream-encode errors. - HIGH-2: add Rust unit test streaming_response_omits_done_when_upstream_returns_invalid_json_mid_stream that drains raw response bytes and asserts: * !contains("data: [DONE]") — no terminal sentinel after error * contains("event: error") — error event emitted * data line parses as JSON with `error` key — envelope shape This pins the contract at the byte level, independent of any SDK parser behavior. - MEDIUM-2: bump streaming-edges-e2e disconnect case `eventDelayMs` 50→200 to match the peer client-abort case. Verification: - cargo test -p aisix-proxy --lib: 139/139 passing (incl. new test) - cargo clippy -p aisix-proxy --tests -- -D warnings: clean - cargo fmt --check: clean
…d APIError from SyntaxError Audit MEDIUM-2 follow-up on PR #198: The prior `surfacedError === true` assertion was satisfied by EITHER: - The new typed `APIError` path — OpenAI Node SDK throws `APIError` after parsing the SSE error frame's JSON envelope - The OLD plain-string path — SDK calls `JSON.parse(sse.data)` BEFORE checking `sse.event === "error"`, so a plain-string payload yields `SyntaxError("Could not parse message into JSON: ...")` A regression that re-introduced `err.to_string()` raw on the SSE frame would still pass the prior assertion. Tighten by capturing the thrown error's class and message and asserting: - `errCtor !== "SyntaxError"` — must not be the SDK parser-error path - `!errMessage.includes("Could not parse message into JSON")` — must not be the SDK parser-error message This pins the user-visible contract that callers see a typed `APIError`, not a spurious parser failure, on abnormal stream termination.
moonming
force-pushed
the
fix/177-streaming-abnormal-termination
branch
from
May 10, 2026 05:55
8ac2f67 to
c2e9189
Compare
This was referenced May 10, 2026
Closed
moonming
added a commit
that referenced
this pull request
May 10, 2026
* fix(proxy): run output guardrails on streaming responses (closes #204) ## Problem Pre-fix the streaming path (`build_sse_stream` in `crates/aisix-proxy/src/chat.rs`) skipped output guardrails entirely. A `kind: "keyword"` deny-list configured on `hook_point: "output"` fired correctly on non-streaming responses but was trivially bypassable by setting `stream: true` — the gateway forwarded SSE chunks verbatim and never accumulated the assistant text for guardrail evaluation. So a forbidden literal in a streamed completion reached the caller untouched. A guardrail that only fires when someone forgets to enable a flag isn't a security control. The reference implementation in this category treats streaming output guardrails as a baseline requirement (every streaming chat-completion request goes through the guardrail dispatcher). ## Fix Buffer-then-check at end-of-stream: 1. `build_sse_stream` now takes an optional `StreamGuardrailContext` (a typed wrapper around `Arc<dyn Guardrail>` + the model name for tracing). When set, the stream loop accumulates `chunk.delta.content` into a `String` buffer. 2. After the loop completes cleanly (no upstream error), if a guardrail context is present, synthesize a `ChatResponse` from the accumulated content + the StreamCompletion's tracked usage stats and call `chain.check_output(&resp).await`. 3. On `Block { reason }`: - Mirror #153's wire-level redaction: emit an SSE `event: error` frame with the OpenAI envelope shape (`error.type: "content_filter"`, generic `"response blocked by content policy"` message — no matched-literal leak). - Set `errored = true` so the terminal `[DONE]` is suppressed (per docs §5 abnormal-termination contract). - Emit `tracing::warn!` with the rich verdict reason (operator- side log carries the matched detail; wire-side does not). - Set `comp.guardrail_blocked = true` so the post-stream telemetry callback records the block on `usage_events`. The call site in `dispatch_streaming` always passes `Some(ctx)` — the chain itself short-circuits when no policies are configured, so the per-chunk overhead for guardrail-free deployments is just a single `Option::is_some()` check + an unused `String` allocation. ## Trade-off (documented) Per-chunk evaluation is wrong for blocking guardrails: by the time chunk N matches the forbidden literal, chunks 1..N-1 have already been emitted — the secret leaks regardless. Buffer-then-check trades latency-to-first-completion for the security guarantee. Streaming masking-style guardrails (where partial leakage is repaired by rewriting tokens) could use a different cadence; that's a v2 concern. ## Tests **Rust unit (`crates/aisix-proxy/src/lib.rs`)** — new `streaming_output_guardrail_blocks_with_sse_error_event_and_no_done` test that drains raw response bytes through the full proxy stack and asserts: - `!wire.contains("data: [DONE]")` — terminal sentinel suppressed - `wire.contains("event: error")` — error frame emitted - error-frame data parses as JSON with `error.type === "content_filter"` - error message equals the redacted static string - error message does NOT contain the matched literal **E2E (`tests/e2e/src/cases/guardrail-output-e2e.test.ts`)** — added streaming case to the existing output-guardrail describe block. Sets up a separate streaming upstream + Model + caller key sharing the existing env-wide guardrail policy. Sends `stream: true` request via raw `fetch`, asserts the wire-level shape (no `[DONE]`, `event: error` present, OpenAI envelope, redacted message, no forbidden literal in the error envelope, upstream IS hit per buffer-then-check semantics). ## Verification - `cargo test -p aisix-proxy --lib`: **158/158 passing** (incl. new test) - `cargo clippy --workspace --lib --tests -- -D warnings`: clean - `cargo fmt --check`: clean - `pnpm tsc --noEmit` (e2e): clean ## References - Issue: #204 (`bug` + `vulnerability`) - `docs/api-proxy.md` §5 (abnormal-termination contract — same SSE shape this fix uses for guardrail blocks) - #153 (non-streaming guardrail redaction — same wire-level contract mirrored to streaming) - #198 (`error_frame_payload` helper — reused here) * fix(streaming-guardrail): audit follow-ups — Bypass telemetry, fast-path skip, Allow e2e Audit fixes on PR #222: - HIGH-1: Output `Bypass` verdict at end-of-stream was silently dropped from telemetry. Pre-fix the streaming arm only matched `Block`; `Bypass { reason }` fell through to `[DONE]` with no side effect, while the non-streaming path captures Bypass into `usage_events.bypass_reason` so operators can audit which policy fail-opened. Added `bypass_reason: String` field to `StreamCompletion`, populated on the Bypass arm with first-bypass- wins semantics. The on_complete closure merges with the input- side `bypass_reason_for_telem` snapshot before emitting telemetry. - MEDIUM-2: No fast-path skip when the guardrail chain is empty — `String` allocated per chunk on every streaming request even for guardrail-free deployments. Added `is_empty(&self) -> bool` default-method to the `Guardrail` trait (returns `false`, preserving safe behavior for custom impls); `GuardrailChain` overrides to return `self.guardrails.is_empty()`. The streaming call site in `chat.rs::dispatch_streaming` now passes `Some(StreamGuardrailContext)` only when the chain is non-empty; guardrail-free deployments skip both the per-chunk content accumulation and the post-loop synthesized-ChatResponse build. - MEDIUM-3: E2E did not cover the streaming-Allow path — a regression that ALWAYS blocks streaming would have passed the existing block-case test. Added a companion case in `guardrail-output-e2e.test.ts` with a separate clean upstream + Model + caller that emits content NOT containing `FORBIDDEN_WORD`. Asserts the wire shape: terminal `[DONE]` IS present, no `event: error`, full assistant content reaches the caller. - MEDIUM-1 (client-disconnect bypasses guardrail check) → filed as #223. The post-loop check sits after `while let Some(item) = upstream.next().await`; client disconnect drops the generator at the last yield suspension, so the post-loop code never executes and `comp.guardrail_blocked` stays `false` even when the buffered content would have blocked. Telemetry under-counts disconnects-as-blocks. Out of scope for this PR (already a step forward vs pre-fix "every streaming request bypasses the guardrail"); recommended fix shape is to spawn a detached task from the `CompleteOnDrop` Drop impl. - LOW-1, LOW-2 (cosmetic — model_name field, message-content clone): reviewed, deferred. Both are micro-optimizations on a path already gated behind the is_empty() fast-path, so they don't affect the dominant guardrail-free deployment. Verification: - cargo test -p aisix-proxy --lib: 158/158 passing - cargo test -p aisix-guardrails --lib: 42/42 passing - cargo clippy --workspace --lib --tests -- -D warnings: clean - cargo fmt --check: clean - pnpm tsc --noEmit (e2e): clean
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Closes #177.
Problem
Per gateway docs §5 (`docs/api-proxy.md`):
The proxy's `build_sse_stream` (`crates/aisix-proxy/src/chat.rs`) correctly emitted an SSE `event: error` chunk on upstream errors — but then unconditionally appended `data: [DONE]\n\n` afterward. SDK consumers that key off `[DONE]` as the clean-completion signal got tricked into treating truncated streams as complete.
Fix
Track whether an error event has been yielded during the stream loop; only emit the terminal `[DONE]` on clean completion. The `on_complete` token-accounting callback still fires unconditionally so TPM/usage telemetry isn't lost on the abnormal-termination path.
```rust
let mut errored = false;
while let Some(item) = upstream.next().await {
let ev = match item {
Ok(chunk) => { ... }
Err(err) => {
errored = true;
Event::default().event("error").data(err.to_string())
}
};
yield Ok::<, Infallible>(ev);
}
on_complete(total_tokens);
if !errored {
yield Ok::<, Infallible>(Event::default().data("[DONE]"));
}
```
Tests
E2E regression test (`tests/e2e/src/cases/streaming-edges-e2e.test.ts`): restored "upstream disconnects mid-stream" case from the held-back queue. Mock upstream emits 2 SSE chunks then closes the TCP connection (no `[DONE]`). Strict assertions per docs §5:
`eventDelayMs: 50` between mock writes ensures both chunks flush to the gateway BEFORE `res.destroy()` aborts the connection. Without the delay, write buffering on the mock side races the destroy and the gateway sometimes sees zero chunks (which would surface as a request-time 502 instead of an iterator error, masking the regression test's intent).
Verification
Audit follow-ups (commits 98dba67, 8ac2f67)
Round 1 (commit 98dba67) — three findings addressed
MEDIUM-1 → fixed: SSE `event: error` frames now carry OpenAI-envelope JSON (`{"error": {"message", "type"}}`) instead of a plain string. The OpenAI Node SDK calls `JSON.parse(sse.data)` BEFORE checking `sse.event`, so the previous plain-string payload would surface as a `SyntaxError` instead of the typed `APIError` callers expect. Error-type taxonomy is sourced from the existing `BridgeError::error_type()` mapping for upstream errors and `internal_error` for fatal-stream-encode errors. See `error_frame_payload` helper in `crates/aisix-proxy/src/chat.rs`.
HIGH-2 → fixed: added Rust unit test `streaming_response_omits_done_when_upstream_returns_invalid_json_mid_stream` (`crates/aisix-proxy/src/lib.rs`) that drains raw response bytes through the full proxy stack and asserts at the byte level:
This pins the docs §5 contract independent of any SDK parser behavior, complementing the e2e test that exercises the same path through the OpenAI Node SDK.
MEDIUM-2 → fixed: bumped `eventDelayMs: 50 → 200` in the disconnect-case mock to match the peer client-abort case for consistency.
Round 2 (commit 8ac2f67) — e2e tightening + scope justification
MEDIUM-2 (round 2) → fixed: the e2e `surfacedError === true` assertion was satisfied by EITHER the new typed `APIError` path OR the prior plain-string path (`SyntaxError` from `JSON.parse(sse.data)` BEFORE the event-type check). A regression that re-introduced `err.to_string()` raw would still pass. Tightened to assert `errCtor !== "SyntaxError"` and `!errMessage.includes("Could not parse message into JSON")` — pins the user-visible contract that callers see a typed `APIError`, not a parser failure.
HIGH-1, HIGH-2 (round 2) → out-of-scope, filed as enhancement: optional strict-mode error envelope sanitization (OpenAI-conformant taxonomy) #199: the audit flagged that `error.message` carries raw `err.to_string()` (which can include upstream provider error text and parser-detail bleed-through) and that `error.type` emits aisix-internal taxonomy (`upstream_decode_error`, `transport_error`, `stream_aborted`, `config_error`) that don't appear in OpenAI's enumerated error.type values. Both leaks pre-exist on the non-streaming path (`crates/aisix-proxy/src/error.rs:127` calls `ProxyError::Bridge(err).to_string()` which transparently forwards to BridgeError's Display, and `error.rs:112` uses `b.error_type()` directly). Fixing only the streaming path here would create divergence; fixing both is out of scope for a streaming-termination PR. Filed as enhancement: optional strict-mode error envelope sanitization (OpenAI-conformant taxonomy) #199 with proposed fix shape (sanitize message + remap type at the proxy boundary, uniformly across both paths).
MEDIUM-3 / LOW-1 / LOW-3 → reviewed, deferred: MEDIUM-3 (use SseDecoder for unit-test parsing) is stylistic — the byte-level `event: error` assertion is the better contract pin since it asserts the actual wire format. LOW-1 (unreachable fallback in `error_frame_payload`) is defensive style; CLAUDE.md "never silently swallow errors" supports keeping the fallback. LOW-3 (PR description correction `ProxyError::error_type` → `BridgeError::error_type`) corrected above.
Out of scope
Note on full e2e suite
The full e2e suite has an unrelated failure in `weighted-routing-distribution-e2e` — filed as #197 (pre-existing bug on `main` where 100/0 split lands instead of declared 70/30). This PR does not touch routing; the same test fails on pristine main. `streaming-edges-e2e` (the only test this PR cares about) passes cleanly.
Summary by CodeRabbit
Bug Fixes
Tests