From 6e5cf0ce451ad6fa6f65184b603154d85f817df1 Mon Sep 17 00:00:00 2001 From: Hyunsik Kang Date: Thu, 13 Aug 2026 07:53:10 +0900 Subject: [PATCH] Do not release body buffers already handed to the sink BodyState.flush() emits every queued buffer and only clears the queue afterwards, so a cancellation arriving while it emits makes dispose() release buffers whose ownership has already been transferred to the sink. Such a buffer is then released twice: once by the parser, and once by the downstream consumer or the discard hook. With Netty, body buffers are slices of the inbound buffer, so the second release frees the inbound buffer prematurely, which surfaces as IllegalReferenceCountException: refCnt: 0, decrement: 1 io.netty.handler.codec.http.DefaultHttpContent.release reactor.netty.channel.FluxReceive.drainReceiver when reactor-netty releases its own share right after onNext. Remove each buffer from the queue before emitting it, mirroring what enqueue() already does, so that dispose() only ever releases buffers the parser still owns. Signed-off-by: Hyunsik Kang --- .../http/codec/multipart/MultipartParser.java | 14 ++++--- .../codec/multipart/MultipartParserTests.java | 41 +++++++++++++++++++ 2 files changed, 50 insertions(+), 5 deletions(-) diff --git a/spring-web/src/main/java/org/springframework/http/codec/multipart/MultipartParser.java b/spring-web/src/main/java/org/springframework/http/codec/multipart/MultipartParser.java index a286ba7283e..24df791d90f 100644 --- a/spring-web/src/main/java/org/springframework/http/codec/multipart/MultipartParser.java +++ b/spring-web/src/main/java/org/springframework/http/codec/multipart/MultipartParser.java @@ -602,13 +602,17 @@ final class MultipartParser extends BaseSubscriber { emit.forEach(buffer -> MultipartParser.this.emitBody(buffer, false)); } + /** + * Emit all queued buffers, removing each from the queue before emitting it so + * that {@link #dispose()} cannot release a buffer that was already handed over + * to the sink. A cancellation arriving while this method emits would otherwise + * release such a buffer a second time. + */ private void flush() { - for (Iterator iterator = this.queue.iterator(); iterator.hasNext(); ) { - DataBuffer buffer = iterator.next(); - boolean last = !iterator.hasNext(); - MultipartParser.this.emitBody(buffer, last); + DataBuffer buffer; + while ((buffer = this.queue.poll()) != null) { + MultipartParser.this.emitBody(buffer, this.queue.isEmpty()); } - this.queue.clear(); } @Override diff --git a/spring-web/src/test/java/org/springframework/http/codec/multipart/MultipartParserTests.java b/spring-web/src/test/java/org/springframework/http/codec/multipart/MultipartParserTests.java index 1d9a5e5d17d..6cd79714b3a 100644 --- a/spring-web/src/test/java/org/springframework/http/codec/multipart/MultipartParserTests.java +++ b/spring-web/src/test/java/org/springframework/http/codec/multipart/MultipartParserTests.java @@ -25,6 +25,8 @@ import reactor.core.publisher.BaseSubscriber; import reactor.core.publisher.Flux; import org.springframework.core.io.buffer.DataBuffer; +import org.springframework.core.io.buffer.DataBufferUtils; +import org.springframework.core.io.buffer.PooledDataBuffer; import org.springframework.core.testfixture.io.buffer.AbstractLeakCheckingTests; import static java.nio.charset.StandardCharsets.UTF_8; @@ -69,4 +71,43 @@ class MultipartParserTests extends AbstractLeakCheckingTests { assertThat(received).singleElement().isInstanceOf(MultipartParser.HeadersToken.class); } + @Test // gh-37115 + void cancelWhileEmittingBodyTokensKeepsEmittedBuffersAllocated() { + byte[] boundary = "simple-boundary".getBytes(UTF_8); + String content = "--simple-boundary\r\nContent-Type: text/plain\r\n\r\n" + + "a".repeat(1024) + "\r\n" + + "--simple-boundary\r\nContent-Type: text/plain\r\n\r\n" + + "b".repeat(1024) + "\r\n--simple-boundary--\r\n"; + byte[] bytes = content.getBytes(UTF_8); + DataBuffer buffer = this.bufferFactory.allocateBuffer(bytes.length); + buffer.write(bytes); + + Flux tokens = MultipartParser.parse(Flux.just(buffer), boundary, 8192, UTF_8); + + List receivedBuffers = new ArrayList<>(); + BaseSubscriber subscriber = new BaseSubscriber<>() { + @Override + protected void hookOnSubscribe(Subscription subscription) { + request(Long.MAX_VALUE); + } + @Override + protected void hookOnNext(MultipartParser.Token token) { + if (token instanceof MultipartParser.BodyToken bodyToken) { + receivedBuffers.add(bodyToken.buffer()); + cancel(); + } + } + }; + tokens.subscribe(subscriber); + + // The cancellation above arrives while the parser emits its queued body buffers. + // Ownership of an emitted buffer belongs to the sink, so the parser must not + // release it on disposal: with Netty, body buffers are slices of the inbound + // buffer, and releasing one twice releases the inbound buffer prematurely. + assertThat(receivedBuffers).isNotEmpty(); + assertThat(receivedBuffers).allSatisfy(received -> + assertThat(((PooledDataBuffer) received).isAllocated()).isTrue()); + receivedBuffers.forEach(DataBufferUtils::release); + } + }