[ISSUE #8195] Bound server-produced SSE queues and slow-consumer session lifetime - #8350

Fixes #8195.

Problem

The SSE transport of the MCP server plugin produces server-initiated JSON-RPC responses and notifications through Flux.create(...) with Reactor's default buffering strategy. When the SSE client stops reading, Reactor Netty stops requesting downstream, and every sink.next(...) call keeps buffering events in an unbounded internal queue: a single slow or dead consumer can grow the gateway heap without limit. The completion of sendMessage only signals submission into that buffer, so a tool response is indistinguishable from one that was actually written, and nothing ever reclaims a session whose consumer stopped reading — the session, its budget-free buffer and its exchange-holder entry live until the client disconnects.

Both public SSE entry points (handleSseConnection and createSseFlux) carried a copy of the same unbounded stream setup, so any fix had to land twice or the gap would remain in one of them.

Approach

Both entry points now share one newSessionStream() implementation: each subscription owns exactly one MCP session, one admission state and one stream sink, wrapped in Flux.defer so per-subscription state can never leak across connections.

Admission is bounded per session by a SseSessionPolicy (defaults: 256 pending events, 1 MiB pending serialized UTF-8 bytes, 30s output-stall timeout). sendMessage serializes the payload, reserves one event slot plus its exact UTF-8 byte length, and only then hands the event to the stream. The reservation is released by a release hook when the consumer pulls the event out of the transport buffer — that buffer is the only unbounded segment of the pipeline, because downstream writes are already bounded by Reactor Netty's socket-writability water marks. Reservations and releases stay FIFO with the event order, so byte accounting is exact without tagging individual events.

Rejections are explicit, never silent drops: a message larger than the whole byte budget is rejected (OVERSIZED) while the session stays usable for others; a message that does not fit the remaining budget (OVERFLOW) means the consumer stopped making progress, so the session is terminated and the failing send returns an error to the MCP SDK instead of pretending success.

A stall detector runs only while output is pending: the first reserved event schedules a check, every consumed event refreshes the progress timestamp, and a session whose pending output makes no progress within the configured bound is terminated. A healthy idle session without pending output is never a candidate.

Termination is idempotent and exactly once: an atomic transition on the session state removes the session and its exchange-holder entry, records a bounded terminal reason (client disconnected, output stalled, output overflow, provider closed) and completes the stream. Completing the stream closes this transport, which is the SDK's own definition of closing the session — McpServerSession.closeGracefully() delegates to the transport — so outstanding operations of the session finish through their own request timeouts and are never replayed. The dispose path, the stall path, overflow rejection and provider shutdown all funnel into the same transition, so concurrent close attempts cannot clean up twice. Completion of the stream respects normal backpressure: buffered events drain in order for a consumer that resumes reading, and the SSE framing, event names and initial endpoint event are unchanged.

Low-cardinality counters (accepted, rejected, and per-reason terminals) are exposed on the provider for diagnostics and tests; neither counters nor log lines carry session identifiers or message content.

The manager accepts an optional policy and passes it to every SSE transport it creates; deployments that do not configure anything get the documented defaults. Configuration exposure through ShenyuConfig/yaml is deliberately out of scope for this change — the policy object is the extension point.

Testing

New ShenyuSseServerTransportProviderTest (11 cases) drives the real transport with captured SDK sessions: a controlled zero-demand subscription proves the finite overflow outcome (bounded acceptance, deterministic rejection, session cleanup, ordered drain and completion once the consumer resumes); byte-budget cases prove serialized UTF-8 accounting on exact boundaries including a multi-byte payload whose character count matches but whose byte count does not; an oversized single message is rejected without terminating the session; the endpoint event is delivered first with order preserved; an idle session survives the stall window while pending output without progress terminates within the bound; a stalled session does not affect a concurrent healthy one; client disconnect releases all state exactly once. Loopback tests over Reactor Netty cover the handleSseConnection entry point end to end: a fast reader receives the endpoint event plus ordered messages, a throttled reader recovers under temporary pressure without loss, and an abrupt disconnect removes the server session. SseSessionPolicyTest locks the documented defaults and validation.

mvn -pl shenyu-plugin/shenyu-plugin-mcp-server test passes with 240 tests (224 pre-existing + 16 new including the policy tests, timing-sensitive cases re-run three times for stability) and Checkstyle is clean.