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 <cj848@hanmail.net>

spring-web/src/main/java/org/springframework/http/codec/multipart/MultipartParser.java

@@ -602,13 +602,17 @@ final class MultipartParser extends BaseSubscriber<DataBuffer> {
 			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<DataBuffer> 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

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<MultipartParser.Token> tokens = MultipartParser.parse(Flux.just(buffer), boundary, 8192, UTF_8);
+
+		List<DataBuffer> receivedBuffers = new ArrayList<>();
+		BaseSubscriber<MultipartParser.Token> 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);
+	}
+
 }