This is an automated email from the ASF dual-hosted git repository.
Aias00 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/shenyu.git
The following commit(s) were added to refs/heads/master by this push:
new 50a25f53f1 fix(plugin): bound MCP tool-response capture memory (#7646)
(#7671)
50a25f53f1 is described below
commit 50a25f53f13327812b7298a0f6d4c73fbc769d6d
Author: Daming <[email protected]>
AuthorDate: Sat Oct 17 20:47:28 2026 +0800
fix(plugin): bound MCP tool-response capture memory (#7646) (#7671)
The Streamable HTTP MCP response decorator accumulated the whole backend
response in a StringBuilder and decoded every chunk with new String(bytes,
UTF_8). That grew memory without bound and corrupted multi-byte UTF-8
sequences split across chunk boundaries.
Bring it to parity with the SSE variant ShenyuMcpResponseDecorator:
- accumulate raw bytes in a ByteArrayOutputStream and decode UTF-8 exactly
once at completion, preserving multi-byte characters across boundaries;
- enforce a 1 MiB capture budget with deterministic over-limit behaviour
(completeExceptionally) instead of returning a truncated result;
- always release the incoming buffer, including the over-limit path;
- make future completion idempotent and share it between writeWith /
writeAndFlushWith / setComplete via finishResponse().
Add tests for cross-chunk UTF-8 decoding and over-limit failure.
Co-authored-by: Luke Haochao Zhuang <[email protected]>
---
.../NonCommittingMcpResponseDecorator.java | 85 +++++++++++++++++-----
.../NonCommittingMcpResponseDecoratorTest.java | 38 ++++++++++
2 files changed, 105 insertions(+), 18 deletions(-)
diff --git
a/shenyu-plugin/shenyu-plugin-mcp-server/src/main/java/org/apache/shenyu/plugin/mcp/server/response/NonCommittingMcpResponseDecorator.java
b/shenyu-plugin/shenyu-plugin-mcp-server/src/main/java/org/apache/shenyu/plugin/mcp/server/response/NonCommittingMcpResponseDecorator.java
index 489ea64ca5..ca607f6c14 100644
---
a/shenyu-plugin/shenyu-plugin-mcp-server/src/main/java/org/apache/shenyu/plugin/mcp/server/response/NonCommittingMcpResponseDecorator.java
+++
b/shenyu-plugin/shenyu-plugin-mcp-server/src/main/java/org/apache/shenyu/plugin/mcp/server/response/NonCommittingMcpResponseDecorator.java
@@ -29,6 +29,8 @@ import
org.springframework.http.server.reactive.ServerHttpResponseDecorator;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
+import java.io.ByteArrayOutputStream;
+import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.CompletableFuture;
import java.util.Objects;
@@ -53,13 +55,32 @@ public class NonCommittingMcpResponseDecorator extends
ServerHttpResponseDecorat
private static final String PLACEHOLDER_SUFFIX = "}";
+ /**
+ * Upper bound, in bytes, of the captured backend response for a single
MCP tool result.
+ * The captured result must be retained in full because the response
future and the
+ * template contract both require a complete result, so the budget keeps
memory bounded
+ * instead of growing with the upstream body. Mirrors the SSE variant
+ * {@link ShenyuMcpResponseDecorator} so both protocols share the same
protection.
+ */
+ private static final int MAX_CAPTURE_BYTES = 1024 * 1024;
+
private final String sessionId;
private final CompletableFuture<String> responseFuture;
private final JsonObject responseTemplate;
- private final StringBuilder body = new StringBuilder();
+ /**
+ * Accumulates the raw response bytes instead of per-chunk strings, so
multi-byte UTF-8
+ * sequences that span chunk boundaries are decoded correctly exactly once
at completion.
+ */
+ private final ByteArrayOutputStream body = new ByteArrayOutputStream();
+
+ /**
+ * Set when the accumulated response exceeds {@link #MAX_CAPTURE_BYTES};
further chunks are
+ * dropped and the response future is completed exceptionally with a
deterministic error.
+ */
+ private volatile boolean captureLimitExceeded;
/**
* Constructs a new non-committing MCP response decorator.
@@ -87,7 +108,7 @@ public class NonCommittingMcpResponseDecorator extends
ServerHttpResponseDecorat
return Flux.from(body)
.doOnDiscard(DataBuffer.class, DataBufferUtils::release)
.doOnNext(this::appendBuffer)
- .doOnComplete(this::completeResponseFuture)
+ .doOnComplete(this::finishResponse)
.then()
.doOnSuccess(aVoid -> LOG.debug("Successfully completed
writeWith for session: {}", sessionId))
.doOnError(error -> handleProcessingError("writeWith", error));
@@ -101,7 +122,7 @@ public class NonCommittingMcpResponseDecorator extends
ServerHttpResponseDecorat
.concatMap(Flux::from)
.doOnDiscard(DataBuffer.class, DataBufferUtils::release)
.doOnNext(this::appendBuffer)
- .doOnComplete(this::completeResponseFuture)
+ .doOnComplete(this::finishResponse)
.then()
.doOnSuccess(aVoid -> LOG.debug("Successfully completed
writeAndFlushWith for session: {}", sessionId))
.doOnError(error -> handleProcessingError("writeAndFlushWith",
error));
@@ -110,25 +131,43 @@ public class NonCommittingMcpResponseDecorator extends
ServerHttpResponseDecorat
@Override
public Mono<Void> setComplete() {
LOG.debug("Completing response for session: {}", sessionId);
- if (!responseFuture.isDone()) {
- responseFuture.complete(processResponse(""));
- }
+ finishResponse();
// Non-committing: do not write to the underlying response.
return Mono.empty();
}
/**
- * Appends a data buffer chunk to the accumulated response body as it is
emitted,
+ * Appends a data buffer chunk to the accumulated response bytes as it is
emitted,
* so that chunks are processed incrementally instead of being buffered as
a list first.
*
+ * <p>The raw bytes are retained (never decoded per chunk) so that
multi-byte UTF-8
+ * sequences spanning chunk boundaries stay intact until the single decode
at completion.
+ * Once {@link #MAX_CAPTURE_BYTES} would be exceeded, remaining chunks are
dropped and the
+ * over-limit state is recorded for deterministic failure. The buffer is
always released.
+ *
* @param buffer the emitted data buffer chunk
*/
private void appendBuffer(final DataBuffer buffer) {
try {
- final byte[] bytes = new byte[buffer.readableByteCount()];
- buffer.read(bytes);
synchronized (body) {
- body.append(new String(bytes, StandardCharsets.UTF_8));
+ if (captureLimitExceeded) {
+ return;
+ }
+ try (DataBuffer.ByteBufferIterator iterator =
buffer.readableByteBuffers()) {
+ while (iterator.hasNext()) {
+ final ByteBuffer byteBuffer =
iterator.next().asReadOnlyBuffer();
+ final int remainingCapacity = MAX_CAPTURE_BYTES -
body.size();
+ if (byteBuffer.remaining() > remainingCapacity) {
+ captureLimitExceeded = true;
+ LOG.warn("Response body for session {} exceeds the
{} byte capture limit",
+ sessionId, MAX_CAPTURE_BYTES);
+ return;
+ }
+ final byte[] bytes = new byte[byteBuffer.remaining()];
+ byteBuffer.get(bytes);
+ body.write(bytes, 0, bytes.length);
+ }
+ }
}
} finally {
DataBufferUtils.release(buffer);
@@ -136,17 +175,27 @@ public class NonCommittingMcpResponseDecorator extends
ServerHttpResponseDecorat
}
/**
- * Completes the response future with the accumulated response data after
all chunks
- * have been received.
- *
- * <p>Applies response template transformations if configured, and
completes the response
- * future with the processed result.
+ * Completes the response future with the accumulated response data after
all chunks have
+ * been received. Decodes the accumulated bytes exactly once, applies the
response template
+ * and completes the response future. Completing is idempotent: if the
future is already
+ * done, this is a no-op. When the capture limit was exceeded the future
is completed
+ * exceptionally with a deterministic error instead of returning a
truncated result.
*/
- private void completeResponseFuture() {
- final String responseBody;
+ private void finishResponse() {
+ if (responseFuture.isDone()) {
+ return;
+ }
+ if (captureLimitExceeded) {
+ responseFuture.completeExceptionally(
+ new IllegalStateException("MCP response exceeds the 1 MiB
capture limit"));
+ return;
+ }
+ final byte[] responseBytes;
synchronized (body) {
- responseBody = body.toString();
+ responseBytes = body.toByteArray();
}
+ // Single decode across the whole payload preserves UTF-8 sequences
split across chunks.
+ final String responseBody = new String(responseBytes,
StandardCharsets.UTF_8);
LOG.debug("Captured response data for session {}, length: {} chars",
sessionId, responseBody.length());
try {
diff --git
a/shenyu-plugin/shenyu-plugin-mcp-server/src/test/java/org/apache/shenyu/plugin/mcp/server/response/NonCommittingMcpResponseDecoratorTest.java
b/shenyu-plugin/shenyu-plugin-mcp-server/src/test/java/org/apache/shenyu/plugin/mcp/server/response/NonCommittingMcpResponseDecoratorTest.java
index 3287a7d49e..57571d0cc5 100644
---
a/shenyu-plugin/shenyu-plugin-mcp-server/src/test/java/org/apache/shenyu/plugin/mcp/server/response/NonCommittingMcpResponseDecoratorTest.java
+++
b/shenyu-plugin/shenyu-plugin-mcp-server/src/test/java/org/apache/shenyu/plugin/mcp/server/response/NonCommittingMcpResponseDecoratorTest.java
@@ -173,6 +173,44 @@ public final class NonCommittingMcpResponseDecoratorTest {
assertEquals(0, buffer.getNativeBuffer().refCnt());
}
+ // ---------------------------------------------------------------
+ // Bounded capture: UTF-8 boundaries and over-limit behaviour
+ // ---------------------------------------------------------------
+
+ @Test
+ void testWriteWithPreservesUtf8AcrossChunkBoundaries() throws Exception {
+ final CompletableFuture<String> future = new CompletableFuture<>();
+ final NonCommittingMcpResponseDecorator decorator =
+ new NonCommittingMcpResponseDecorator(delegate, "session-1",
future, null);
+
+ // "a€b" => 0x61 0xE2 0x82 0xAC 0x62; split inside the 3-byte euro
sign.
+ final byte[] full = "a€b".getBytes(StandardCharsets.UTF_8);
+ final DataBuffer first = bufferFactory.wrap(new byte[]{full[0],
full[1]});
+ final DataBuffer second = bufferFactory.wrap(new byte[]{full[2],
full[3], full[4]});
+
+ decorator.writeWith(Flux.just(first, second)).block();
+
+ assertEquals("a€b", future.get(5, TimeUnit.SECONDS));
+ }
+
+ @Test
+ void testWriteWithExceedingCaptureLimitFailsDeterministically() {
+ final int overLimitBytes = 1024 * 1024 + 1;
+ final NettyDataBufferFactory factory = new
NettyDataBufferFactory(PooledByteBufAllocator.DEFAULT);
+ final NettyDataBuffer buffer = factory.allocateBuffer(overLimitBytes);
+ buffer.write(new byte[overLimitBytes]);
+
+ final CompletableFuture<String> future = new CompletableFuture<>();
+ final NonCommittingMcpResponseDecorator decorator = new
NonCommittingMcpResponseDecorator(
+ new MockServerHttpResponse(), "session", future, null);
+
+ decorator.writeWith(Mono.just(buffer)).block();
+
+ assertTrue(future.isCompletedExceptionally());
+ // The over-limit buffer must still be released.
+ assertEquals(0, buffer.getNativeBuffer().refCnt());
+ }
+
private DataBuffer buffer(final String content) {
return bufferFactory.wrap(content.getBytes(StandardCharsets.UTF_8));
}
