[Task][MCP Server] Bound server-produced SSE queues and slow-consumer session lifetime #8195

Goal and ownership

Add bounded pending-message/byte admission and a slow-consumer policy to the existing mcpServer SSE session transport. Protect independently produced MCP JSON-RPC responses and notifications when the SSE client consumes slowly or stops reading.

This belongs in the existing MCP transport; it is not a new generic SSE plugin or a change to AI Proxy's encoded-response relay.

Current source evidence

Reviewed source: apache/shenyu@750673e3f619e335333150ff1b2d14070d5d5c31 (2.7.3-SNAPSHOT).

• shenyu-plugin/shenyu-plugin-mcp-server/src/main/java/org/apache/shenyu/plugin/mcp/server/transport/ShenyuSseServerTransportProvider.java creates the response stream with Flux.create(...) without an explicit finite overflow policy; the direct createSseFlux entry at line 295 does the same. Reactor's default create strategy buffers produced values when downstream demand is absent.
• shenyu-plugin/shenyu-plugin-mcp-server/src/main/java/org/apache/shenyu/plugin/mcp/server/transport/ShenyuSseServerTransportProvider.java serializes one JSON-RPC message, calls sink.next(event) and completes its returned Mono. This signals submission to the sink, not socket write progress or a finite pending-byte reservation.
• Both SSE entry points send an initial endpoint event and publish active sessions. Cancellation/disposal currently remove session/exchange-holder entries; graceful provider close completes sessions. Reuse these mechanisms while making active-session ownership explicit.

The transport already participates in WebFlux/Reactor Netty demand. This task identifies a push-producer admission/policy gap; it does not assert a reproduced production OOM or that all MCP/Agent Gateway paths lack backpressure.

Why this is separate

#8148 explicitly excludes independently produced MCP SSE session messages. #8149 covers WebSocket relay; #8150 covers the existing gRPC bridge. #8062 is configuration-revision retirement and #7929 is gateway draining. #7951/#7957 govern resource/business invocation admission and distributed quotas, not this outbound session queue. Keep these scopes distinct and reuse their terminal ownership where relevant.

Implementation scope

1. Apply one session-owned policy to both handleSseConnection and createSseFlux. Preserve the initial endpoint event before JSON-RPC messages and existing SSE serialization, event names and ordering.
2. Bound pending serialized bytes, pending event count and the behavior of one oversized message. Reserve/release budget through actual admission, delivery/discard and terminal paths. Include serializer/operator handoffs in the documented resource bound; a bounded queue alone must not leave unbounded pending producer tasks.
3. Make sendMessage completion/error semantics explicit: bounded acceptance or backpressured delivery must be distinguishable from successful network consumption. If a producer cannot honor demand, reject/cancel within finite limits. Do not silently drop JSON-RPC responses or protocol notifications, or create detached subscriptions per message.
4. Define sustained output-stall detection only while output is pending. A healthy idle session with no messages must not time out as a slow consumer. On overflow/stall, terminate the affected session deterministically; do not promise an error event can reach a peer that stopped reading.
5. Use an idempotent terminal path to cancel timers/producers, release queued data, close the associated SDK session and remove session/exchange-holder state. Specify the treatment of outstanding operations owned by that session; do not transparently replay side-effecting tool calls on reconnect. Session close and configuration/gateway draining must not race into duplicate cleanup.
6. Expose bounded pending-byte/event/stall/terminal-reason metrics. Do not label them with session IDs, raw JSON-RPC IDs, tool arguments or arbitrary endpoint paths. Document configuration defaults, compatibility and protocol-specific limits rather than hard-coding a universal timeout.

Acceptance criteria

• A controlled-demand test holds downstream demand at zero, sends more messages than the configured count/byte limit through the real transport, and proves a finite deterministic outcome. Test the single-event oversize boundary and serialized UTF-8 byte accounting.
• Real loopback SSE tests cover fast, throttled and stopped readers, recovery and abrupt disconnect. One affected session must not stall or close a concurrent healthy session.
• The initial endpoint event is preserved, messages remain ordered, and temporary pressure recovers without truncating, duplicating or silently dropping JSON-RPC messages.
• A long idle session without pending output survives. A session with pending output and no write progress terminates within the configured stall bound; accepted work cannot create an unbounded task backlog.
• Completion, cancellation, overflow, serialization/write failure, provider close and shutdown races release pending budgets/timers and session/exchange-holder state exactly once, with no post-close sends retaining work.
• Both public SSE entry points have equivalent policy coverage; existing normal traffic is compatible. Document reconnect/reinitialization and cancellation semantics without replay or billing guarantees.

Explicit non-goals

Reimplementing all MCP transports, adding Agent Gateway distributed quotas, changing tools authorization, claiming SDK-managed Streamable HTTP has identical buffering, generic SSE reframing, or replacing WebFlux response writes with manual ctx.write() calls. Independently audit other transports before broadening support claims.