tju-yxq opened a new issue, #2208: URL: https://github.com/apache/rocketmq-dashboard/issues/2208
## Problem `OpenAiCompatibleLlmGateway` protects the server from an unbounded cached thread pool by using at most 16 workers and a `SynchronousQueue`, but its rejection policy is `CallerRunsPolicy`. That policy is not safe for the `/api/ai/chat` SSE contract: when all 16 workers are busy, the next long-running HTTP or CLI chat executes on the servlet request thread. The controller cannot return its `SseEmitter` until that provider call completes or times out. A deterministic probe against the current gateway starts 16 blocked streams and invokes a seventeenth `chat()` on another thread. The seventeenth call remains blocked instead of returning an emitter; it only returns after one of the first streams is released. The existing comment describes this as backpressure, but for a streaming endpoint it turns saturation into request-thread occupation for up to the provider/CLI timeout. The gateway also does not connect the `SseEmitter` lifecycle to the submitted task. No `onCompletion`, `onTimeout`, or `onError` callback cancels queued or running work. If the browser closes the connection or the downstream SSE response times out, the provider request can continue occupying one of the 16 workers until its independent timeout. Repeated abandoned requests can therefore keep the pool saturated even though no client can consume their output. Cancellation by thread interruption already has useful behavior in most paths: `HttpClient.send` and the OpenAI SSE reader surface interruption, and `CliAgentProvider.complete` forcibly destroys its child process. Claude's streaming path is an exception: it catches `InterruptedException` after creating the process but does not destroy that process, so prompt-enhancement cancellation can leave a CLI child running. ## Expected behavior - Never execute a long-running chat task on the servlet caller thread. - When the bounded executor is saturated, return the emitter immediately and finish it with a structured `503 / llm.gateway.overloaded` SSE error. - Register completion, timeout, and error callbacks before or safely around task attachment, and cancel the corresponding `Future` when the downstream client is gone. - Handle the race where an emitter terminates before the submitted `Future` is attached. - Treat cancellation as a normal lifecycle outcome: do not attempt to send another error to a closed emitter and do not log it as an upstream provider failure. - Preserve the existing HTTP/CLI timeouts as upper bounds, while allowing downstream disconnects to release work earlier. - Destroy the Claude streaming process when its worker is interrupted, matching the non-streaming CLI cleanup contract. - Keep incomplete-config and unsupported-provider error emitters independent of the saturated chat executor. ## Proposed implementation Introduce a small SSE chat-session abstraction that owns one emitter and one submitted task. It should register the emitter callbacks, attach/cancel the task atomically, expose cancellation state to send/error paths, and ensure terminal events are emitted at most once. Replace `CallerRunsPolicy` with explicit rejection and convert `RejectedExecutionException` into the overload SSE contract without scheduling more work on the same saturated executor. For deterministic tests, allow the gateway to use a package-visible injected executor/emitter factory while retaining the existing production limits. The complete lifecycle fix is expected to exceed 100 lines of production code because it needs a reusable session state machine, explicit submission/rejection handling, integration across HTTP and CLI chat paths, and Claude process cleanup; the size should not come from duplicated branches. ## Test coverage Add tests that prove: 1. with every worker blocked, another `chat()` returns promptly, no provider work runs on the test/caller thread, and the returned emitter receives the structured overload error; 2. emitter completion, timeout, and error each cancel a running task, including the attach-after-termination race; 3. a cancelled HTTP stream is interrupted and does not attempt a second terminal response; 4. normal success and provider failure still produce exactly one terminal sequence; 5. interrupting Claude streaming destroys its child process and preserves the thread interrupt status; 6. gateway shutdown cancels active work. The tests should use latches and controlled test emitters/processes rather than sleeping for provider timeouts or requiring a live LLM/CLI. ## Related work Issue #1564 concerns the deadline while consuming an upstream OpenAI-compatible SSE body. PR #2042 bounds Claude streaming output, and PR #2197 handles provider error envelopes. Those changes do not address servlet-thread execution under saturation or cancellation when the downstream Studio SSE client disconnects. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
