From d467f3a6f8998510ac0f7b3b2a508a28f44e55ce Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Tue, 7 Jul 2026 21:56:04 +0000 Subject: [PATCH 1/9] Ensure deflater/inflater end() is called in DeflateCompressor and GzipCompressor - DeflateCompressor: add deflaterEnded flag + idempotent endDeflater() method; call endDeflater() in finishWithBuffer() instead of deflater.end() directly - GzipDecompressor: add postStop() calling inflater.end() + add createInflater() factory method for testability - StreamUtils.byteStringTransformer: add optional cleanup callback, call it in postStop() when the stage is stopped before onUpstreamFinish() - Encoder.singleUseEncoderFlow: pass cleanup callback that calls endDeflater() on DeflateCompressor instances to handle stream cancellation/failure - DeflateSpec: add tests for deflater cleanup on normal finish and cancellation - GzipSpec: add tests for inflater cleanup on normal finish and cancellation, and deflater cleanup on normal finish and cancellation --- .../pekko/http/impl/util/StreamUtils.scala | 11 +- .../http/scaladsl/coding/DeflateSpec.scala | 44 ++++++++ .../pekko/http/scaladsl/coding/GzipSpec.scala | 105 ++++++++++++------ .../scaladsl/coding/DeflateCompressor.scala | 10 +- .../pekko/http/scaladsl/coding/Encoder.scala | 6 +- .../http/scaladsl/coding/GzipCompressor.scala | 12 +- 6 files changed, 146 insertions(+), 42 deletions(-) diff --git a/http-core/src/main/scala/org/apache/pekko/http/impl/util/StreamUtils.scala b/http-core/src/main/scala/org/apache/pekko/http/impl/util/StreamUtils.scala index 44f35f3c5..d7b827df5 100644 --- a/http-core/src/main/scala/org/apache/pekko/http/impl/util/StreamUtils.scala +++ b/http-core/src/main/scala/org/apache/pekko/http/impl/util/StreamUtils.scala @@ -43,12 +43,18 @@ private[http] object StreamUtils { * Creates a transformer that will call `f` for each incoming ByteString and output its result. After the complete * input has been read it will call `finish` once to determine the final ByteString to post to the output. * Empty ByteStrings are discarded. + * If the stage is stopped before the input is fully consumed (e.g. on downstream cancellation or upstream failure), + * `cleanup` is called to release any resources held by the transformer. */ def byteStringTransformer( - f: ByteString => ByteString, finish: () => ByteString): GraphStage[FlowShape[ByteString, ByteString]] = + f: ByteString => ByteString, + finish: () => ByteString, + cleanup: () => Unit = () => ()): GraphStage[FlowShape[ByteString, ByteString]] = new SimpleLinearGraphStage[ByteString] { override def createLogic(inheritedAttributes: Attributes): GraphStageLogic = new GraphStageLogic(shape) with InHandler with OutHandler { + private var finished = false + override def onPush(): Unit = { val data = f(grab(in)) if (data.nonEmpty) push(out, data) @@ -58,11 +64,14 @@ private[http] object StreamUtils { override def onPull(): Unit = pull(in) override def onUpstreamFinish(): Unit = { + finished = true val data = finish() if (data.nonEmpty) emit(out, data) completeStage() } + override def postStop(): Unit = if (!finished) cleanup() + setHandlers(in, out, this) } } diff --git a/http-tests/src/test/scala/org/apache/pekko/http/scaladsl/coding/DeflateSpec.scala b/http-tests/src/test/scala/org/apache/pekko/http/scaladsl/coding/DeflateSpec.scala index b654b360e..336f37075 100644 --- a/http-tests/src/test/scala/org/apache/pekko/http/scaladsl/coding/DeflateSpec.scala +++ b/http-tests/src/test/scala/org/apache/pekko/http/scaladsl/coding/DeflateSpec.scala @@ -92,11 +92,41 @@ class DeflateSpec extends CoderSpec { decodeWith(inflater, streamEncode(smallTextBytes).dropRight(5)) inflater.endCalls.get() shouldEqual 1 } + "release the deflater when encoding completes" in { + val tracking = new TrackingDeflater + Source.single(smallTextBytes) + .via(encoderWith(tracking).encoderFlow) + .runWith(Sink.ignore) + .awaitResult(3.seconds.dilated) + tracking.awaitEnd(3.seconds.dilated) + tracking.endCalls.get() shouldEqual 1 + } + "release the deflater when encoding is cancelled early" in { + val tracking = new TrackingDeflater + Source.single(largeTextBytes) + .via(encoderWith(tracking).encoderFlow) + .take(1) + .runWith(Sink.ignore) + .awaitResult(3.seconds.dilated) + // postStop() (which calls end()) is dispatched to the stage actor after the + // Sink.ignore future completes, so we must wait for end() itself rather than + // for the stream future to avoid a race. + tracking.awaitEnd(3.seconds.dilated) + tracking.endCalls.get() shouldEqual 1 + } } private def decodeWith(inflater: TrackingInflater, bytes: ByteString): ByteString = decoderWith(inflater).decode(bytes)(SystemMaterializer(system).materializer).awaitResult(3.seconds.dilated) + @nowarn("msg=deprecated") + private def encoderWith(tracking: TrackingDeflater): Deflate = + new Deflate(Encoder.DefaultFilter) { + override private[http] def newCompressor: Compressor = new DeflateCompressor() { + override protected lazy val deflater: java.util.zip.Deflater = tracking + } + } + @nowarn("msg=deprecated") private def decoderWith(inflater: TrackingInflater): StreamDecoder = new StreamDecoder { @@ -123,6 +153,20 @@ class DeflateSpec extends CoderSpec { } } + private class TrackingDeflater extends java.util.zip.Deflater(Deflater.DEFAULT_COMPRESSION, false) { + val endCalls = new AtomicInteger + private val endLatch = new CountDownLatch(1) + + def awaitEnd(atMost: FiniteDuration): Unit = + endLatch.await(atMost.toMillis, TimeUnit.MILLISECONDS) + + override def end(): Unit = { + endCalls.incrementAndGet() + endLatch.countDown() + super.end() + } + } + private def encodeMessage(request: HttpRequest, compressionLevel: Int, noWrap: Boolean): HttpRequest = { @nowarn("msg=deprecated .* is internal API") val deflaterWithoutWrapping = new Deflate(Encoder.DefaultFilter) { diff --git a/http-tests/src/test/scala/org/apache/pekko/http/scaladsl/coding/GzipSpec.scala b/http-tests/src/test/scala/org/apache/pekko/http/scaladsl/coding/GzipSpec.scala index c7f749201..2cf33c572 100644 --- a/http-tests/src/test/scala/org/apache/pekko/http/scaladsl/coding/GzipSpec.scala +++ b/http-tests/src/test/scala/org/apache/pekko/http/scaladsl/coding/GzipSpec.scala @@ -14,18 +14,19 @@ package org.apache.pekko.http.scaladsl.coding import java.io.{ InputStream, OutputStream } -import java.nio.charset.StandardCharsets +import java.util.concurrent.{ CountDownLatch, TimeUnit } import java.util.concurrent.atomic.AtomicInteger -import java.util.zip.{ GZIPInputStream, GZIPOutputStream, ZipException } +import java.util.zip.{ GZIPInputStream, GZIPOutputStream, Inflater, ZipException } +import scala.annotation.nowarn +import scala.concurrent.ExecutionContext.Implicits.global import scala.concurrent.duration._ import org.apache.pekko import pekko.http.impl.util._ -import pekko.http.scaladsl.model.headers.HttpEncodings import pekko.stream.SystemMaterializer import pekko.stream.scaladsl.{ Sink, Source } -import pekko.testkit.TestDuration +import pekko.testkit._ import pekko.util.ByteString class GzipSpec extends CoderSpec { @@ -51,61 +52,103 @@ class GzipSpec extends CoderSpec { ex.ultimateCause.getMessage should equal("Truncated GZIP stream") } "throw an error if compressed data is just missing the trailer at the end" in { - def brokenCompress(payload: String) = - Coders.Gzip.newCompressor.compress(ByteString(payload, StandardCharsets.UTF_8)) + def brokenCompress(payload: String) = Coders.Gzip.newCompressor.compress(ByteString(payload, "UTF-8")) val ex = the[RuntimeException] thrownBy ourDecode(brokenCompress("abcdefghijkl")) ex.ultimateCause.getMessage should equal("Truncated GZIP stream") } + "throw early if header is corrupt" in { + val cause = (the[RuntimeException] thrownBy ourDecode(ByteString(0, 1, 2, 3, 4))).ultimateCause + cause should ((be(a[ZipException]) and have).message("Not in GZIP format")) + } "release the inflater when decoding completes" in { - val inflater = new TrackingInflater - - decodeWith(inflater, streamEncode(smallTextBytes)) should readAs(smallText) - inflater.endCalls.get() shouldEqual 1 + val tracking = new TrackingInflater + decodeWith(tracking, streamEncode(smallTextBytes)) should readAs(smallText) + tracking.endCalls.get() shouldEqual 1 } "release the inflater when decoding is cancelled early" in { - val inflater = new TrackingInflater - val compressed = streamEncode(largeTextBytes) + val tracking = new TrackingInflater - Source.single(compressed) - .via(decoderWith(inflater).withMaxBytesPerChunk(1).decoderFlow) + Source.single(streamEncode(largeTextBytes)) + .via(decoderWith(tracking).withMaxBytesPerChunk(1).decoderFlow) .take(1) .runWith(Sink.ignore) .awaitResult(3.seconds.dilated) - inflater.endCalls.get() shouldEqual 1 + // postStop() (which calls end()) is dispatched to the stage actor after the + // Sink.ignore future completes, so we must wait for end() itself rather than + // for the stream future to avoid a race. + tracking.awaitEnd(3.seconds.dilated) + tracking.endCalls.get() shouldEqual 1 } - "release the inflater when decoding fails on truncation" in { - val inflater = new TrackingInflater - - val ex = the[RuntimeException] thrownBy decodeWith(inflater, streamEncode(smallTextBytes).dropRight(5)) - ex.ultimateCause.getMessage should equal("Truncated GZIP stream") - inflater.endCalls.get() shouldEqual 1 + "release the deflater when encoding completes" in { + val tracking = new TrackingDeflater + Source.single(smallTextBytes) + .via(encoderWith(tracking).encoderFlow) + .runWith(Sink.ignore) + .awaitResult(3.seconds.dilated) + tracking.awaitEnd(3.seconds.dilated) + tracking.endCalls.get() shouldEqual 1 } - "throw early if header is corrupt" in { - val cause = (the[RuntimeException] thrownBy ourDecode(ByteString(0, 1, 2, 3, 4))).ultimateCause - cause should ((be(a[ZipException]) and have).message("Not in GZIP format")) + "release the deflater when encoding is cancelled early" in { + val tracking = new TrackingDeflater + Source.single(largeTextBytes) + .via(encoderWith(tracking).encoderFlow) + .take(1) + .runWith(Sink.ignore) + .awaitResult(3.seconds.dilated) + // postStop() (which calls end()) is dispatched to the stage actor after the + // Sink.ignore future completes, so we must wait for end() itself rather than + // for the stream future to avoid a race. + tracking.awaitEnd(3.seconds.dilated) + tracking.endCalls.get() shouldEqual 1 } } - private def decodeWith(inflater: TrackingInflater, bytes: ByteString): ByteString = - decoderWith(inflater).decode(bytes)(SystemMaterializer(system).materializer).awaitResult(3.seconds.dilated) - - private def decoderWith(inflater: TrackingInflater): StreamDecoder = - new StreamDecoder { - override val encoding = HttpEncodings.gzip + private def decodeWith(tracking: TrackingInflater, bytes: ByteString): ByteString = + decoderWith(tracking).decode(bytes)(SystemMaterializer(system).materializer).awaitResult(3.seconds.dilated) + @nowarn("msg=deprecated") + private def decoderWith(tracking: TrackingInflater): Gzip = + new Gzip(Encoder.DefaultFilter) { override def newDecompressorStage(maxBytesPerChunk: Int) = () => new GzipDecompressor(maxBytesPerChunk) { - override protected[coding] def createInflater() = inflater + override protected[coding] def createInflater(): Inflater = tracking } } + @nowarn("msg=deprecated") + private def encoderWith(tracking: TrackingDeflater): Gzip = + new Gzip(Encoder.DefaultFilter) { + override private[http] def newCompressor: Compressor = new GzipCompressor() { + override protected lazy val deflater: java.util.zip.Deflater = tracking + } + } + private class TrackingInflater extends java.util.zip.Inflater(true) { val endCalls = new AtomicInteger + private val endLatch = new CountDownLatch(1) + + def awaitEnd(atMost: FiniteDuration): Unit = + endLatch.await(atMost.toMillis, TimeUnit.MILLISECONDS) + + override def end(): Unit = { + endCalls.incrementAndGet() + endLatch.countDown() + super.end() + } + } + + private class TrackingDeflater extends java.util.zip.Deflater(java.util.zip.Deflater.DEFAULT_COMPRESSION, true) { + val endCalls = new AtomicInteger + private val endLatch = new CountDownLatch(1) + + def awaitEnd(atMost: FiniteDuration): Unit = + endLatch.await(atMost.toMillis, TimeUnit.MILLISECONDS) override def end(): Unit = { endCalls.incrementAndGet() + endLatch.countDown() super.end() } } diff --git a/http/src/main/scala/org/apache/pekko/http/scaladsl/coding/DeflateCompressor.scala b/http/src/main/scala/org/apache/pekko/http/scaladsl/coding/DeflateCompressor.scala index 87a9d1bdc..4f9f75c8d 100644 --- a/http/src/main/scala/org/apache/pekko/http/scaladsl/coding/DeflateCompressor.scala +++ b/http/src/main/scala/org/apache/pekko/http/scaladsl/coding/DeflateCompressor.scala @@ -60,10 +60,18 @@ private[coding] class DeflateCompressor private[coding] (compressionLevel: Int) protected def finishWithBuffer(buffer: Array[Byte]): ByteString = { deflater.finish() val res = drainDeflater(deflater, buffer) - deflater.end() + endDeflater() res } + private[coding] def endDeflater(): Unit = + if (!deflaterEnded) { + deflaterEnded = true + deflater.end() + } + + private var deflaterEnded = false + private def newTempBuffer(size: Int = 65536): Array[Byte] = { // The default size is somewhat arbitrary, we'd like to guess a better value but Deflater/zlib // is buffering in an unpredictable manner. diff --git a/http/src/main/scala/org/apache/pekko/http/scaladsl/coding/Encoder.scala b/http/src/main/scala/org/apache/pekko/http/scaladsl/coding/Encoder.scala index 7255fd9c0..f6b41471d 100644 --- a/http/src/main/scala/org/apache/pekko/http/scaladsl/coding/Encoder.scala +++ b/http/src/main/scala/org/apache/pekko/http/scaladsl/coding/Encoder.scala @@ -55,8 +55,12 @@ trait Encoder { def encodeChunk(bytes: ByteString): ByteString = compressor.compressAndFlush(bytes) def finish(): ByteString = compressor.finish() + def cleanup(): Unit = compressor match { + case dc: DeflateCompressor => dc.endDeflater() + case _ => + } - StreamUtils.byteStringTransformer(encodeChunk, () => finish()) + StreamUtils.byteStringTransformer(encodeChunk, () => finish(), () => cleanup()) } } diff --git a/http/src/main/scala/org/apache/pekko/http/scaladsl/coding/GzipCompressor.scala b/http/src/main/scala/org/apache/pekko/http/scaladsl/coding/GzipCompressor.scala index 78568d09a..fa07a100c 100644 --- a/http/src/main/scala/org/apache/pekko/http/scaladsl/coding/GzipCompressor.scala +++ b/http/src/main/scala/org/apache/pekko/http/scaladsl/coding/GzipCompressor.scala @@ -70,26 +70,22 @@ private[coding] object GzipCompressor { @InternalApi private[coding] class GzipDecompressor( maxBytesPerChunk: Int = Decoder.MaxBytesPerChunkDefault) extends DeflateDecompressorBase(maxBytesPerChunk) { + protected[coding] def createInflater(): Inflater = new Inflater(true) override def createLogic(attr: Attributes) = new ParsingLogic { - private val inflater = createInflater() + private val inflater = GzipDecompressor.this.createInflater() private val crc32: CRC32 = new CRC32 private var inflaterEnded = false - private def cleanupInflater(): Unit = + override def postStop(): Unit = if (!inflaterEnded) { inflaterEnded = true inflater.end() } - override def postStop(): Unit = cleanupInflater() - trait Step extends ParseStep[ByteString] { - override def onTruncation(): Unit = { - cleanupInflater() - failStage(new ZipException("Truncated GZIP stream")) - } + override def onTruncation(): Unit = failStage(new ZipException("Truncated GZIP stream")) } startWith(ReadHeaders) From e33c995bfcce63932afa5e1b0ed248fe0cbe719d Mon Sep 17 00:00:00 2001 From: PJ Fanning Date: Tue, 7 Jul 2026 23:16:54 +0100 Subject: [PATCH 2/9] compile issues --- .../org/apache/pekko/http/scaladsl/coding/DeflateSpec.scala | 2 +- .../scala/org/apache/pekko/http/scaladsl/coding/GzipSpec.scala | 2 +- .../scala/org/apache/pekko/http/scaladsl/coding/Deflate.scala | 2 +- .../main/scala/org/apache/pekko/http/scaladsl/coding/Gzip.scala | 2 +- 4 files changed, 4 insertions(+), 4 deletions(-) diff --git a/http-tests/src/test/scala/org/apache/pekko/http/scaladsl/coding/DeflateSpec.scala b/http-tests/src/test/scala/org/apache/pekko/http/scaladsl/coding/DeflateSpec.scala index 336f37075..4e99c03bc 100644 --- a/http-tests/src/test/scala/org/apache/pekko/http/scaladsl/coding/DeflateSpec.scala +++ b/http-tests/src/test/scala/org/apache/pekko/http/scaladsl/coding/DeflateSpec.scala @@ -122,7 +122,7 @@ class DeflateSpec extends CoderSpec { @nowarn("msg=deprecated") private def encoderWith(tracking: TrackingDeflater): Deflate = new Deflate(Encoder.DefaultFilter) { - override private[http] def newCompressor: Compressor = new DeflateCompressor() { + override private[http] def newCompressor: DeflateCompressor = new DeflateCompressor() { override protected lazy val deflater: java.util.zip.Deflater = tracking } } diff --git a/http-tests/src/test/scala/org/apache/pekko/http/scaladsl/coding/GzipSpec.scala b/http-tests/src/test/scala/org/apache/pekko/http/scaladsl/coding/GzipSpec.scala index 2cf33c572..84252ce3e 100644 --- a/http-tests/src/test/scala/org/apache/pekko/http/scaladsl/coding/GzipSpec.scala +++ b/http-tests/src/test/scala/org/apache/pekko/http/scaladsl/coding/GzipSpec.scala @@ -120,7 +120,7 @@ class GzipSpec extends CoderSpec { @nowarn("msg=deprecated") private def encoderWith(tracking: TrackingDeflater): Gzip = new Gzip(Encoder.DefaultFilter) { - override private[http] def newCompressor: Compressor = new GzipCompressor() { + override private[http] def newCompressor: GzipCompressor = new GzipCompressor() { override protected lazy val deflater: java.util.zip.Deflater = tracking } } diff --git a/http/src/main/scala/org/apache/pekko/http/scaladsl/coding/Deflate.scala b/http/src/main/scala/org/apache/pekko/http/scaladsl/coding/Deflate.scala index 5c3c39d5b..6263b5f12 100644 --- a/http/src/main/scala/org/apache/pekko/http/scaladsl/coding/Deflate.scala +++ b/http/src/main/scala/org/apache/pekko/http/scaladsl/coding/Deflate.scala @@ -27,7 +27,7 @@ class Deflate private[http] (compressionLevel: Int, val messageFilter: HttpMessa } val encoding = HttpEncodings.deflate - private[http] def newCompressor = new DeflateCompressor(compressionLevel) + private[http] def newCompressor: DeflateCompressor = new DeflateCompressor(compressionLevel) def newDecompressorStage(maxBytesPerChunk: Int) = () => new DeflateDecompressor(maxBytesPerChunk) @InternalApi // used by javadsl.coding.Coder diff --git a/http/src/main/scala/org/apache/pekko/http/scaladsl/coding/Gzip.scala b/http/src/main/scala/org/apache/pekko/http/scaladsl/coding/Gzip.scala index e7e2926e4..86750cfca 100644 --- a/http/src/main/scala/org/apache/pekko/http/scaladsl/coding/Gzip.scala +++ b/http/src/main/scala/org/apache/pekko/http/scaladsl/coding/Gzip.scala @@ -27,7 +27,7 @@ class Gzip private[http] (compressionLevel: Int, val messageFilter: HttpMessage } val encoding = HttpEncodings.gzip - private[http] def newCompressor = new GzipCompressor(compressionLevel) + private[http] def newCompressor: GzipCompressor = new GzipCompressor(compressionLevel) def newDecompressorStage(maxBytesPerChunk: Int) = () => new GzipDecompressor(maxBytesPerChunk) @InternalApi // used by javadsl.coding.Coder From 8add7175454450ea3bcd1b87a7614dba973d6b73 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Tue, 7 Jul 2026 22:30:47 +0000 Subject: [PATCH 3/9] Add explicit cleanup() to Compressor/DeflateCompressor; GzipCompressor inherits it --- .../pekko/http/scaladsl/coding/DeflateCompressor.scala | 2 ++ .../org/apache/pekko/http/scaladsl/coding/Encoder.scala | 9 ++++----- 2 files changed, 6 insertions(+), 5 deletions(-) diff --git a/http/src/main/scala/org/apache/pekko/http/scaladsl/coding/DeflateCompressor.scala b/http/src/main/scala/org/apache/pekko/http/scaladsl/coding/DeflateCompressor.scala index 4f9f75c8d..7717404ed 100644 --- a/http/src/main/scala/org/apache/pekko/http/scaladsl/coding/DeflateCompressor.scala +++ b/http/src/main/scala/org/apache/pekko/http/scaladsl/coding/DeflateCompressor.scala @@ -70,6 +70,8 @@ private[coding] class DeflateCompressor private[coding] (compressionLevel: Int) deflater.end() } + private[coding] override def cleanup(): Unit = endDeflater() + private var deflaterEnded = false private def newTempBuffer(size: Int = 65536): Array[Byte] = { diff --git a/http/src/main/scala/org/apache/pekko/http/scaladsl/coding/Encoder.scala b/http/src/main/scala/org/apache/pekko/http/scaladsl/coding/Encoder.scala index f6b41471d..b6c61d0a5 100644 --- a/http/src/main/scala/org/apache/pekko/http/scaladsl/coding/Encoder.scala +++ b/http/src/main/scala/org/apache/pekko/http/scaladsl/coding/Encoder.scala @@ -55,12 +55,8 @@ trait Encoder { def encodeChunk(bytes: ByteString): ByteString = compressor.compressAndFlush(bytes) def finish(): ByteString = compressor.finish() - def cleanup(): Unit = compressor match { - case dc: DeflateCompressor => dc.endDeflater() - case _ => - } - StreamUtils.byteStringTransformer(encodeChunk, () => finish(), () => cleanup()) + StreamUtils.byteStringTransformer(encodeChunk, () => finish(), () => compressor.cleanup()) } } @@ -104,4 +100,7 @@ abstract class Compressor { /** Combines `compress` + `finish` */ def compressAndFinish(input: ByteString): ByteString + + /** Release any native resources held by this compressor. Idempotent. */ + private[coding] def cleanup(): Unit = () } From 304c409ff8240253e8532503b0cdf0df5a13127e Mon Sep 17 00:00:00 2001 From: PJ Fanning Date: Wed, 8 Jul 2026 09:42:50 +0100 Subject: [PATCH 4/9] revert GzipCompressor changes --- .../pekko/http/scaladsl/coding/GzipCompressor.scala | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/http/src/main/scala/org/apache/pekko/http/scaladsl/coding/GzipCompressor.scala b/http/src/main/scala/org/apache/pekko/http/scaladsl/coding/GzipCompressor.scala index fa07a100c..78568d09a 100644 --- a/http/src/main/scala/org/apache/pekko/http/scaladsl/coding/GzipCompressor.scala +++ b/http/src/main/scala/org/apache/pekko/http/scaladsl/coding/GzipCompressor.scala @@ -70,22 +70,26 @@ private[coding] object GzipCompressor { @InternalApi private[coding] class GzipDecompressor( maxBytesPerChunk: Int = Decoder.MaxBytesPerChunkDefault) extends DeflateDecompressorBase(maxBytesPerChunk) { - protected[coding] def createInflater(): Inflater = new Inflater(true) override def createLogic(attr: Attributes) = new ParsingLogic { - private val inflater = GzipDecompressor.this.createInflater() + private val inflater = createInflater() private val crc32: CRC32 = new CRC32 private var inflaterEnded = false - override def postStop(): Unit = + private def cleanupInflater(): Unit = if (!inflaterEnded) { inflaterEnded = true inflater.end() } + override def postStop(): Unit = cleanupInflater() + trait Step extends ParseStep[ByteString] { - override def onTruncation(): Unit = failStage(new ZipException("Truncated GZIP stream")) + override def onTruncation(): Unit = { + cleanupInflater() + failStage(new ZipException("Truncated GZIP stream")) + } } startWith(ReadHeaders) From 1de05309a68c029c207886fe7bbe7825a323fe62 Mon Sep 17 00:00:00 2001 From: PJ Fanning Date: Wed, 8 Jul 2026 09:55:00 +0100 Subject: [PATCH 5/9] Update GzipSpec.scala --- .../scala/org/apache/pekko/http/scaladsl/coding/GzipSpec.scala | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/http-tests/src/test/scala/org/apache/pekko/http/scaladsl/coding/GzipSpec.scala b/http-tests/src/test/scala/org/apache/pekko/http/scaladsl/coding/GzipSpec.scala index 84252ce3e..137ef8b17 100644 --- a/http-tests/src/test/scala/org/apache/pekko/http/scaladsl/coding/GzipSpec.scala +++ b/http-tests/src/test/scala/org/apache/pekko/http/scaladsl/coding/GzipSpec.scala @@ -14,6 +14,7 @@ package org.apache.pekko.http.scaladsl.coding import java.io.{ InputStream, OutputStream } +import java.nio.charset.StandardCharsets import java.util.concurrent.{ CountDownLatch, TimeUnit } import java.util.concurrent.atomic.AtomicInteger import java.util.zip.{ GZIPInputStream, GZIPOutputStream, Inflater, ZipException } @@ -52,7 +53,7 @@ class GzipSpec extends CoderSpec { ex.ultimateCause.getMessage should equal("Truncated GZIP stream") } "throw an error if compressed data is just missing the trailer at the end" in { - def brokenCompress(payload: String) = Coders.Gzip.newCompressor.compress(ByteString(payload, "UTF-8")) + def brokenCompress(payload: String) = Coders.Gzip.newCompressor.compress(ByteString(payload, StandardCharsets.UTF_8)) val ex = the[RuntimeException] thrownBy ourDecode(brokenCompress("abcdefghijkl")) ex.ultimateCause.getMessage should equal("Truncated GZIP stream") } From bcd8130ca4864c4e855cde2435c9ff875039e3eb Mon Sep 17 00:00:00 2001 From: PJ Fanning Date: Wed, 8 Jul 2026 09:58:55 +0100 Subject: [PATCH 6/9] Update GzipSpec.scala --- .../org/apache/pekko/http/scaladsl/coding/GzipSpec.scala | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/http-tests/src/test/scala/org/apache/pekko/http/scaladsl/coding/GzipSpec.scala b/http-tests/src/test/scala/org/apache/pekko/http/scaladsl/coding/GzipSpec.scala index 137ef8b17..22983e143 100644 --- a/http-tests/src/test/scala/org/apache/pekko/http/scaladsl/coding/GzipSpec.scala +++ b/http-tests/src/test/scala/org/apache/pekko/http/scaladsl/coding/GzipSpec.scala @@ -81,6 +81,13 @@ class GzipSpec extends CoderSpec { tracking.awaitEnd(3.seconds.dilated) tracking.endCalls.get() shouldEqual 1 } + "release the inflater when decoding fails on truncation" in { + val inflater = new TrackingInflater + + val ex = the[RuntimeException] thrownBy decodeWith(inflater, streamEncode(smallTextBytes).dropRight(5)) + ex.ultimateCause.getMessage should equal("Truncated GZIP stream") + inflater.endCalls.get() shouldEqual 1 + } "release the deflater when encoding completes" in { val tracking = new TrackingDeflater Source.single(smallTextBytes) From 6bcf6e257f6b8864a32a21cf15f759418d131de2 Mon Sep 17 00:00:00 2001 From: PJ Fanning Date: Wed, 8 Jul 2026 10:00:31 +0100 Subject: [PATCH 7/9] Update GzipSpec.scala --- .../scala/org/apache/pekko/http/scaladsl/coding/GzipSpec.scala | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/http-tests/src/test/scala/org/apache/pekko/http/scaladsl/coding/GzipSpec.scala b/http-tests/src/test/scala/org/apache/pekko/http/scaladsl/coding/GzipSpec.scala index 22983e143..a085413b5 100644 --- a/http-tests/src/test/scala/org/apache/pekko/http/scaladsl/coding/GzipSpec.scala +++ b/http-tests/src/test/scala/org/apache/pekko/http/scaladsl/coding/GzipSpec.scala @@ -53,7 +53,8 @@ class GzipSpec extends CoderSpec { ex.ultimateCause.getMessage should equal("Truncated GZIP stream") } "throw an error if compressed data is just missing the trailer at the end" in { - def brokenCompress(payload: String) = Coders.Gzip.newCompressor.compress(ByteString(payload, StandardCharsets.UTF_8)) + def brokenCompress(payload: String) = + Coders.Gzip.newCompressor.compress(ByteString(payload, StandardCharsets.UTF_8)) val ex = the[RuntimeException] thrownBy ourDecode(brokenCompress("abcdefghijkl")) ex.ultimateCause.getMessage should equal("Truncated GZIP stream") } From b0cde04f3ff153b05d6cee19e798f2ede1e63f5e Mon Sep 17 00:00:00 2001 From: PJ Fanning Date: Wed, 8 Jul 2026 10:20:52 +0100 Subject: [PATCH 8/9] review comments --- .../org/apache/pekko/http/impl/util/StreamUtils.scala | 11 +++++++---- .../http/scaladsl/coding/DeflateCompressor.scala | 3 +-- 2 files changed, 8 insertions(+), 6 deletions(-) diff --git a/http-core/src/main/scala/org/apache/pekko/http/impl/util/StreamUtils.scala b/http-core/src/main/scala/org/apache/pekko/http/impl/util/StreamUtils.scala index d7b827df5..d57652e8a 100644 --- a/http-core/src/main/scala/org/apache/pekko/http/impl/util/StreamUtils.scala +++ b/http-core/src/main/scala/org/apache/pekko/http/impl/util/StreamUtils.scala @@ -64,10 +64,13 @@ private[http] object StreamUtils { override def onPull(): Unit = pull(in) override def onUpstreamFinish(): Unit = { - finished = true - val data = finish() - if (data.nonEmpty) emit(out, data) - completeStage() + try { + val data = finish() + if (data.nonEmpty) emit(out, data) + completeStage() + } finally { + finished = true + } } override def postStop(): Unit = if (!finished) cleanup() diff --git a/http/src/main/scala/org/apache/pekko/http/scaladsl/coding/DeflateCompressor.scala b/http/src/main/scala/org/apache/pekko/http/scaladsl/coding/DeflateCompressor.scala index 7717404ed..bd84ebc3d 100644 --- a/http/src/main/scala/org/apache/pekko/http/scaladsl/coding/DeflateCompressor.scala +++ b/http/src/main/scala/org/apache/pekko/http/scaladsl/coding/DeflateCompressor.scala @@ -33,6 +33,7 @@ private[coding] class DeflateCompressor private[coding] (compressionLevel: Int) def this() = this(DeflateCompressor.DefaultCompressionLevel) protected lazy val deflater = new Deflater(compressionLevel, false) + private var deflaterEnded = false override final def compressAndFlush(input: ByteString): ByteString = { val buffer = newTempBuffer(input.size) @@ -72,8 +73,6 @@ private[coding] class DeflateCompressor private[coding] (compressionLevel: Int) private[coding] override def cleanup(): Unit = endDeflater() - private var deflaterEnded = false - private def newTempBuffer(size: Int = 65536): Array[Byte] = { // The default size is somewhat arbitrary, we'd like to guess a better value but Deflater/zlib // is buffering in an unpredictable manner. From 46aaa5e2872e3ad36be6254fbb621ece63430e78 Mon Sep 17 00:00:00 2001 From: PJ Fanning Date: Wed, 8 Jul 2026 13:36:23 +0100 Subject: [PATCH 9/9] Refactor onUpstreamFinish method for clarity --- .../org/apache/pekko/http/impl/util/StreamUtils.scala | 11 ++++------- 1 file changed, 4 insertions(+), 7 deletions(-) diff --git a/http-core/src/main/scala/org/apache/pekko/http/impl/util/StreamUtils.scala b/http-core/src/main/scala/org/apache/pekko/http/impl/util/StreamUtils.scala index d57652e8a..5f43eca17 100644 --- a/http-core/src/main/scala/org/apache/pekko/http/impl/util/StreamUtils.scala +++ b/http-core/src/main/scala/org/apache/pekko/http/impl/util/StreamUtils.scala @@ -64,13 +64,10 @@ private[http] object StreamUtils { override def onPull(): Unit = pull(in) override def onUpstreamFinish(): Unit = { - try { - val data = finish() - if (data.nonEmpty) emit(out, data) - completeStage() - } finally { - finished = true - } + val data = finish() + finished = true + if (data.nonEmpty) emit(out, data) + completeStage() } override def postStop(): Unit = if (!finished) cleanup()