This is an automated email from the ASF dual-hosted git repository.
pjfanning pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko-http.git
The following commit(s) were added to refs/heads/main by this push:
new a06ad5738 GzipDecompressor: fix inflater memory leak (#1134)
a06ad5738 is described below
commit a06ad57386088fbf1bf9a18ba10bec67eae55385
Author: PJ Fanning <[email protected]>
AuthorDate: Wed Jul 8 09:28:45 2026 +0100
GzipDecompressor: fix inflater memory leak (#1134)
* Initial plan
* Apply remaining changes
* Update GzipSpec.scala
* Update GzipCompressor.scala
---------
Co-authored-by: copilot-swe-agent[bot]
<[email protected]>
---
.../pekko/http/scaladsl/coding/GzipSpec.scala | 59 +++++++++++++++++++++-
.../http/scaladsl/coding/GzipCompressor.scala | 18 ++++++-
2 files changed, 74 insertions(+), 3 deletions(-)
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 ada29ac59..c7f749201 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,10 +14,18 @@
package org.apache.pekko.http.scaladsl.coding
import java.io.{ InputStream, OutputStream }
+import java.nio.charset.StandardCharsets
+import java.util.concurrent.atomic.AtomicInteger
import java.util.zip.{ GZIPInputStream, GZIPOutputStream, ZipException }
+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.util.ByteString
class GzipSpec extends CoderSpec {
@@ -43,13 +51,62 @@ 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")
}
+ "release the inflater when decoding completes" in {
+ val inflater = new TrackingInflater
+
+ decodeWith(inflater, streamEncode(smallTextBytes)) should
readAs(smallText)
+ inflater.endCalls.get() shouldEqual 1
+ }
+ "release the inflater when decoding is cancelled early" in {
+ val inflater = new TrackingInflater
+ val compressed = streamEncode(largeTextBytes)
+
+ Source.single(compressed)
+ .via(decoderWith(inflater).withMaxBytesPerChunk(1).decoderFlow)
+ .take(1)
+ .runWith(Sink.ignore)
+ .awaitResult(3.seconds.dilated)
+
+ inflater.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
+ }
"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"))
}
}
+
+ 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
+
+ override def newDecompressorStage(maxBytesPerChunk: Int) =
+ () =>
+ new GzipDecompressor(maxBytesPerChunk) {
+ override protected[coding] def createInflater() = inflater
+ }
+ }
+
+ private class TrackingInflater extends java.util.zip.Inflater(true) {
+ val endCalls = new AtomicInteger
+
+ override def end(): Unit = {
+ endCalls.incrementAndGet()
+ super.end()
+ }
+ }
}
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 ab8b07ab5..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,12 +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 = new Inflater(true)
+ private val inflater = createInflater()
private val crc32: CRC32 = new CRC32
+ private var inflaterEnded = false
+
+ 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)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
