Add server-side WebSocket compression - #1114
Conversation
| * Dependency versions have been updated | ||
| * Jackson3 is supported in a new pekko-http-jackson3 lib | ||
| * Changed Java DSL methods that return Scala Durations to return Java Durations ([PR788](https://github.com/apache/pekko-http/pull/788)) | ||
| * Server-side WebSocket connections can now negotiate RFC 7692 `permessage-deflate` compression when requested by the client. |
There was a problem hiding this comment.
2.0.0-M1 is already released. Just revert this and we will form 2.0.0-M2 release notes when that release is ready, hopefully in a month or 2.
|
I see netty have many compression support, should we do that too? |
| # Whether the server should support WebSocket compression using the RFC 7692 | ||
| # permessage-deflate extension. Compression is negotiated during the | ||
| # WebSocket handshake and is only used when the client requests it. | ||
| enabled = true |
There was a problem hiding this comment.
Should we do adaptive compression?
There was a problem hiding this comment.
Could you clarify what kind of adaptive compression you mean?
With this PR, Pekko HTTP already supports adaptation in two ways:
- Compression is only negotiated when the client offers
permessage-deflate. - Compression can be disabled globally, or per accepted WebSocket via the new
compressionEnabledoverloads.
If you mean adapting per outbound message, for example compressing a large message but sending a small message uncompressed, I think that could make sense as a follow-up PR. permessage-deflate allows this: once the extension is negotiated, a sender can still choose to leave RSV1 unset for individual messages. This would need careful handling around fragmented messages and context takeover, but a minimum-size threshold or application-controlled filter could be useful.
In Play we do basically the same for the gzip filter: https://github.com/playframework/playframework/blob/d87116f880789e0dbc40b7f971e08638e5ab7557/web/play-filters-helpers/src/main/resources/reference.conf#L297-L300 - which avoids gziping content that is smaller then a certain threshold.
But, as said, I would do that in a follow up PR if desired.
Similarly, skipping compression for certain message types or already-compressed payloads could make sense later, but Pekko HTTP currently does not know application-level message semantics at this layer, so that would also need separate design.
If you mean changing the compression level dynamically, that is more complicated and probably less useful as a general automatic feature. A simpler and more realistic variant would be allowing applications to choose a compression level per WebSocket connection, similar to how this PR allows enabling/disabling compression per accepted WebSocket. That should be much easier to reason about than changing the level message-by-message.
I would also do that in a follow up PR if desired.
If you mean disabling compression automatically under CPU or memory pressure, I think that is much larger in scope and probably not a good fit for this PR. It would need metrics, policy, and predictable behavior, and I am not sure the added complexity would be worth it.
There was a problem hiding this comment.
Did some research if netty has similiar feature(s):
What Netty does have is lower-level extension filters:
WebSocketExtensionFilter.mustSkip(WebSocketFrame frame)Those filters can decide per frame whether an extension encoder/decoder should be skipped. Netty’s own tests show a threshold-style example:
return (frame instanceof TextWebSocketFrame || frame instanceof BinaryWebSocketFrame)
&& frame.content().readableBytes() < 100;So Netty supports the mechanism for "don’t compress small messages", but it does not ship a named adaptive compression feature/config. Its default is WebSocketExtensionFilterProvider.DEFAULT, which means NEVER_SKIP, so negotiated messages are compressed.
Important detail: for permessage-deflate, Netty forbids changing the decision in the middle of a fragmented message. If compression/decompression is already in progress and the filter suddenly says “skip” for a continuation frame, Netty throws.
So yeah, maybe in a follow up PR it might make sense to think about some mechanism to dynamicaly decide if an outbound message should be compressed or not
There was a problem hiding this comment.
yes, but that can come up later, we are using something like this later.
There was a problem hiding this comment.
I have now opened follow-up #1172, which adds application-controlled per-message compression selection. The compression decision is evaluated once per message and retained across continuation frames, including when control frames are interleaved.
For this PR, no. This PR should stay focused on RFC 7692 |
5af04ba to
50c0cde
Compare
|
All fixed. For full disclosure: I was working on this for like 1,5 days intensively with heavy use of codex (gpt-5.5), doing lots of rubberducking, analysing how netty or projects like tomcat implements things and going over the RFC. Did many dozens iterations, back and forth (trying different approaches). Zero code copied, some manual adjustments. Since the netty implementation is different - by having different architecture in general - and Akka HTTP does not even have support for compressed websocket, this code here is unique ("my own work"). Also, made sure that this will play nicely together with Play - where I locally build support for websocket compression as well based on this implementation. Will submit the Play PR soon so you see what I mean. |
|
|
||
| /** | ||
| * Java API | ||
| */ |
There was a problem hiding this comment.
can you add @since 2.0.0 here?
50c0cde to
c6e2be2
Compare
pjfanning
left a comment
pjfanning
left a comment
There was a problem hiding this comment.
lgtm - thanks
Hopefully, we can make some of the suggested improvements over the next few weeks but I think this is already a very useful addition as is.
* Negotiate RFC 7692 permessage-deflate for server-side WebSockets. * Add compression settings, documentation, release notes, and MiMa filters. * Cover negotiation, compression, fragmentation, context takeover, max allocation, and low-level frame handling.
* Add WebSocketUpgrade overloads that accept a compressionEnabled flag. * Wire accepted WebSockets to decline negotiated compression per request. * Document and test per-upgrade compression disabling.
c6e2be2 to
d626ab3
Compare
|
(only rebased on latest main with last force push) |
He-Pin
left a comment
He-Pin
left a comment
There was a problem hiding this comment.
Nice work on this PR — server-side WebSocket compression is a long-requested feature and the implementation looks solid overall. The negotiation logic is thorough, the test coverage is impressive (fragmentation, context takeover, max-allocation, low-level frames), and the per-upgrade compression control is a nice touch.
A few things I noticed while reading through the code:
PerMessageDeflate.scala
-
Buffer allocation in
inflate/deflate: The 1024-byte buffer is allocated on every call toinflate()/deflate(). Since these are called per-frame within a single connection, it might be worth making the buffer a class-level field inInflaterFlow/DeflaterFlowto reduce allocation pressure. Also, 1024 bytes is quite small — 8192 is a more typical buffer size for streaming decompression/compression and would reduce loop iterations for larger messages. -
CompressedFrame.appendandUncompressedFrame.append: Usingdata ++ nexton ByteString for each incoming fragment creates a new ByteString each time, which can become O(n²) for heavily fragmented messages. Consider collecting fragments in aVector[ByteString]and concatenating once infinishFrame(), or using aByteStringBuilder. -
asInstanceOf[WebSocketSettingsImpl]inHandshake.scala: The unchecked castsettings.asInstanceOf[WebSocketSettingsImpl].compressionwill throw aClassCastExceptionif a customWebSocketSettingsimplementation is provided. Consider using a pattern match with a fallback (e.g. skip compression negotiation for unknown settings types) or documenting the requirement. -
validWindowBitsedge case: The function doesn't guard against an empty string explicitly. While the empty-string case forclient_max_window_bitsis handled beforevalidWindowBitsis called, the function itself would throwNumberFormatExceptionon"".toIntif ever called with an empty value in the future. A simplevalue.nonEmpty &&guard would make it more defensive. -
LifecycleMapConcatStage— incomplete message on upstream finish: If upstream finishes while a fragmented compressed message is still in progress (compressedMessageInProgress = trueorcompressedFrame.isDefined), the pending data is silently dropped. Consider whether this should emit a protocol error or at least log a warning.
Configuration
- Default
max-allocation = 64k: 64 KB is quite conservative for a default decompressed message size limit. Many real-world WebSocket applications (e.g. real-time collaboration, gaming, large JSON payloads) routinely exceed this. While it's a security default and can be overridden, a value like 1 MB or 256 KB might be more practical and still provide protection against decompression bombs. Tomcat defaults to no limit, and Jetty defaults to 128 KB for the text message size limit.
Minor / Style
-
Header list construction in
buildResponse: TheList(...) ::: perMessageDeflate.map(...).toList ::: List(...)pattern is a bit awkward. ASeq.newBuilderor a conditional++might read more cleanly. -
WebSocketCompressionSettingsImpl.Disabledsentinel: TheDisabledvalue hasmaxAllocation = 0, which the reference.conf documents as "no limit". While this value is never used when compression is disabled, the semantic mismatch is slightly confusing.
The test coverage is excellent — especially the fragmented compressed messages with interleaved control frames, the context takeover reset tests, and the max-allocation boundary tests. Overall this is a well-structured PR.
He-Pin
left a comment
He-Pin
left a comment
There was a problem hiding this comment.
Left a few inline suggestions on the core implementation files.
| val input = if (appendTail) data ++ EmptyStoredBlock else data | ||
| inflater.setInput(input.toArray) | ||
| val output = new ByteArrayOutputStream() | ||
| val buffer = new Array[Byte](1024) |
There was a problem hiding this comment.
The 1024-byte buffer is allocated fresh on every inflate() call. Since this method runs per frame within a single connection, it might be worth making this a class-level field to reduce GC pressure. Also, 1024 is pretty small — something like 8192 would cut down loop iterations for larger messages without meaningfully increasing memory usage.
There was a problem hiding this comment.
The 1024-byte buffer is allocated fresh on every inflate() call. Since this method runs per frame within a single connection, it might be worth making this a class-level field to reduce GC pressure.
codex was suggesting the same, but we came to the conclusion its not worth it. anyway, let me see again.
There was a problem hiding this comment.
in the Artery Transport, there is a bytebufPoll. I think that would help.
There was a problem hiding this comment.
Addressed by #1173. It uses one reusable 8 KiB byte array owned by each materialized InflaterFlow, rather than a shared pool. Since each flow is per connection and processed serially, this removes the per-call allocation without adding pool acquire/release overhead.
| private def deflate(data: ByteString, removeTail: Boolean): ByteString = { | ||
| deflater.setInput(data.toArray) | ||
| val output = new ByteArrayOutputStream() | ||
| val buffer = new Array[Byte](1024) |
There was a problem hiding this comment.
Same as above — this buffer could be a field of DeflaterFlow and sized to 8192 or so. The deflater loop will spin many more times than necessary on larger payloads with a 1KB buffer.
There was a problem hiding this comment.
Addressed by #1173: DeflaterFlow now owns and reuses an 8 KiB buffer instead of allocating a new 1 KiB buffer for every deflate() call.
|
|
||
| private final case class CompressedFrame(header: FrameHeader, data: ByteString, appendTail: Boolean) { | ||
| def append(next: ByteString): CompressedFrame = copy(data = data ++ next) | ||
| } |
There was a problem hiding this comment.
'data ++ next creates a new ByteString copy on every fragment. For heavily fragmented messages this becomes O(n²) in the total payload size. Consider accumulating fragments in a Vector[ByteString] (or a ByteStringBuilder) and concatenating once in finishFrame(). Same applies to UncompressedFrame.append below.'
There was a problem hiding this comment.
Addressed by #1173: fragments are accumulated in a Vector[ByteString] and compacted once using a size-hinted ByteStringBuilder. The single-fragment case returns the existing ByteString directly.
| } | ||
|
|
||
| private def validWindowBits(value: String): Boolean = | ||
| value.length <= 2 && value.forall(_.isDigit) && { |
There was a problem hiding this comment.
Nit: if this ever gets called with an empty string (it won't today because the empty case is handled before reaching here), value.toInt would throw NumberFormatException. A quick value.nonEmpty && guard at the start would make this more defensive for future callers.
There was a problem hiding this comment.
Addressed by #1173 by adding the defensive value.nonEmpty check before parsing the window-bits value.
| override def onPull(): Unit = | ||
| pushOrPull() | ||
|
|
||
| override def onUpstreamFinish(): Unit = { |
There was a problem hiding this comment.
If upstream finishes while a fragmented compressed message is still in progress (compressedMessageInProgress == true or compressedFrame.isDefined), the pending data is silently dropped here. Not sure if that's the right behavior — maybe emit a protocol error or at least log a warning so it's not a silent data loss?
There was a problem hiding this comment.
#1173 now verifies that inflater and deflater resources are released even when the stream finishes with incomplete message state. It intentionally leaves the resulting wire and stream semantics unchanged. I plan to address those separately: incomplete inbound compressed state should produce a FrameError and WebSocket close 1002, while incomplete outbound state should fail the stream stage.
There was a problem hiding this comment.
Correction to my previous reply: after comparing this more closely with Netty and general WebSocket shutdown semantics, I no longer plan a compression-specific behavioral change.
Netty releases incomplete encoder and decoder state during handler removal or channel inactivity without inspecting the in-progress compression state or synthesizing a 1002 close. A Close frame may also terminate a connection while a fragmented data message is unfinished, and after a real TCP EOF sending another Close frame is generally impossible.
#1173 addresses the actionable part by verifying that inflater and deflater resources are released for incomplete inbound and outbound state, including incomplete logical messages and physical frames.
Any future change to truncated-message reporting should be implemented in the general WebSocket framing layer and cover compressed and uncompressed fragmented messages consistently.
| val perMessageDeflate = | ||
| PerMessageDeflate.negotiate( | ||
| clientRequestedExtensions, | ||
| settings.asInstanceOf[WebSocketSettingsImpl].compression) |
There was a problem hiding this comment.
This unchecked cast will throw a ClassCastException if someone provides a custom WebSocketSettings implementation that isn't WebSocketSettingsImpl. Might be worth a pattern match with a graceful fallback (e.g. skip compression negotiation for unknown settings types), or at least documenting that WebSocketSettings must be a WebSocketSettingsImpl.
There was a problem hiding this comment.
Addressed by #1173: the handshake now pattern-matches WebSocketSettingsImpl. Unknown custom implementations skip compression negotiation instead of causing a ClassCastException.
| # exceeded while inflating a compressed message, the connection is closed | ||
| # with a WebSocket protocol error. | ||
| # Set to 0 to disable this limit. | ||
| max-allocation = 64k |
There was a problem hiding this comment.
64 KB feels quite conservative for a default. Real-world WebSocket apps (collaborative editing, gaming, large JSON payloads) routinely exceed this. Tomcat has no limit by default, Jetty uses 128 KB for text message size. Something like 256 KB or 1 MB might be a more practical default while still protecting against decompression bombs. Users who need tighter limits can always lower it.
There was a problem hiding this comment.
I would be ok with 256 KB - 1 MB seems a bit too much.
@pjfanning what do you think?
There was a problem hiding this comment.
I have no strong preference. 256Kb can be the starting point. If we get requests to increase the default later on, we can do that. The value is a configurable but a default setting that only some people need to worry about is ideal.
|
@He-Pin Give me a minute, so my agent can answer your agent ;) |
He-Pin
left a comment
He-Pin
left a comment
There was a problem hiding this comment.
overall lgtm , would it possible to coodinate with apache/pekko#2409 ?
| } | ||
| val clientRequestedExtensions = headers.collect { | ||
| case extensions: `Sec-WebSocket-Extensions` => extensions.extensions | ||
| }.flatten |
There was a problem hiding this comment.
Would it be possible to avoid the flatten and do it in one call?
There was a problem hiding this comment.
Addressed by #1173 by replacing collect(...).flatten with a single flatMap.
| BidiFlow.fromFlows(inflaterFlow, deflaterFlow) | ||
|
|
||
| def frameEventBidiFlow( | ||
| maskRandom: () => Random): BidiFlow[FrameEvent, FrameEvent, FrameEvent, FrameEvent, NotUsed] = |
There was a problem hiding this comment.
This follows the existing WebSocket masking API: WebSocketSettings.randomFactory and Masking currently use java.util.Random. Using RandomGenerator consistently would require a broader WebSocket settings/masking API change rather than a compression-specific change, so I would leave that for a separate PR.
|
|
||
| private trait LifecycleMapConcat[-In, +Out] extends (In => immutable.Iterable[Out]) { | ||
| def close(): Unit | ||
| } |
There was a problem hiding this comment.
I think we should avoid this MapConcat name because there is one in the pekko stream too.
There was a problem hiding this comment.
And there is a statefulMap which can be used
There was a problem hiding this comment.
I looked at both built-in operators. statefulMapConcat only invokes its completion callback on normal upstream completion, so it cannot guarantee that Inflater.end() and Deflater.end() run after upstream failure or downstream cancellation. statefulMap has broader lifecycle handling but is one-to-one, so we would need to emit an Iterable and add another flattening stage.
The private LifecycleMapConcatStage directly supports zero or more outputs and calls close() from postStop, covering completion, failure, cancellation, and protocol errors. #1173 adds tests for all of those termination paths. I therefore kept the custom stage; the Lifecycle prefix is intended to distinguish it from Pekko Stream’s regular mapConcat operator.
| * | ||
| * @since 2.0.0 | ||
| */ | ||
| def handleMessagesWith( |
There was a problem hiding this comment.
The complete method documentation is present on org.apache.pekko.http.javadsl.model.ws.WebSocketUpgrade. The corresponding methods in the Scala DSL trait are bridge implementations of that Java API and follow the existing convention of using the Java API marker. The new public methods also have @since 2.0.0.
2409 is not making progress and we can always add zstd later - I don't see why we can't just use the compression here initially |
|
I'm fine with this, Which I'm asking is will @mkurz introduce some common interface into pekko and used it here. |
|
I will address the comments later or tomorrow |
| output.write(buffer, 0, count) | ||
| count = inflater.inflate(buffer) | ||
| } | ||
| ByteString.fromArray(output.toByteArray) |
There was a problem hiding this comment.
Addressed in the merged implementation: the freshly created output arrays are wrapped using ByteString.fromArrayUnsafe. They are not modified after ownership is transferred to the ByteString. Input uses toArrayUnsafe before being passed to the JDK inflater or deflater, which only reads that input.
| } | ||
|
|
||
| private def finishFrame(): immutable.Iterable[FrameEventOrError] = { | ||
| val frame = compressedFrame.get |
There was a problem hiding this comment.
Netty (DeflateDecoder.java:48-49, 122-123) handles empty EMPTY_DEFLATE_BLOCK
There was a problem hiding this comment.
Addressed in the merged implementation and covered by explicit tests. An empty inbound compressed message is inflated and delivered as an empty message, while an empty outbound message produces the expected DEFLATE block and successfully round-trips through an inflater. Final compressed frames also have the RFC 7692 tail restored before inflation.
| val input = if (appendTail) data ++ EmptyStoredBlock else data | ||
| inflater.setInput(input.toArray) | ||
| val output = new ByteArrayOutputStream() | ||
| val buffer = new Array[Byte](1024) |
There was a problem hiding this comment.
is it possible we reuse this buffer?
There was a problem hiding this comment.
Yes. Addressed by #1173: the buffer is now an 8 KiB field of InflaterFlow and is reused across inflate operations.
|
I think we can merge this, and improvement can come up later, wdyt @pjfanning I checked the netty's implemenation, the current is great and correct. but yes, netty can have better performance. |
|
merged the existing PR - we can mark this as experimental in the next release if there are still issues that need addressing at that point |
3 participants
What changed
Adds server-side support for RFC 7692
permessage-deflateWebSocket compression.Compression is negotiated during the WebSocket handshake when the client offers
permessage-deflateinSec-WebSocket-Extensions. It is enabled by default and can be disabled globally through configuration.Also adds an API option to decline negotiated compression for a single accepted WebSocket upgrade.
Details
This PR adds:
permessage-deflatenegotiation for server-side WebSocketsclient_max_window_bitsWebSocketUpgradeoverloadsNetty comparison
The implementation follows Netty’s
permessage-deflatebehavior where it makes sense for Pekko HTTP:server_no_context_takeoverandclient_no_context_takeovercan be negotiatedclient_max_window_bitscan be negotiatedmax-allocationOne deliberate difference is zlib implementation scope. Netty also uses the JDK
DeflaterandInflaterAPIs for the default zlib settings, and only switches to JZlib when non-default zlibwindowBitsormemLevelsupport is needed. Pekko HTTP currently uses only the JDKDeflaterandInflaterAPIs, like Tomcat and Jetty. The JDK API does not expose zlibwindowBitsormemLevel, so Pekko HTTP does not acceptserver_max_window_bitsvalues below15and does not provide server window-size or memory-level settings.Clients may still request
client_max_window_bits; when they do, Pekko HTTP can include the configuredpreferred-client-window-sizein the handshake response to ask the client to use that window size for client-to-server messages.Security notes
WebSocket compression can increase CPU and memory usage.
The default context takeover settings match common server defaults and favor compression efficiency. Applications that need a more conservative setup can configure no-context-takeover behavior or disable compression globally/per accepted WebSocket.
max-allocationlimits decompressed message size. If the limit is exceeded while inflating a compressed message, Pekko HTTP closes the connection with a WebSocket protocol error.Testing
Ran:
sbt "http-core / Test / testOnly org.apache.pekko.http.impl.engine.ws.WebSocketServerSpec" sbt +mimaReportBinaryIssuesAlso verified Play Framework integration against the locally published Pekko HTTP artifact:
(This is a WIP pull request for Play which I will submit tomorrow - I will you ping you then)
References