fix: bound the size a compressed payload may expand to

[pjfanning]

Motivation

Five serializers gzip their payload and decompress it on the way back in, each with the
same unbounded loop: read the whole GZIPInputStream into a ByteArrayOutputStream and
return toByteArray. gzip expands by up to about three orders of magnitude, so neither
the size of the compressed bytes nor the transport's frame limit bounds the buffer the
decompressed bytes are read into — a payload well inside maximum-frame-size can expand
to hundreds of megabytes.

Two of the twelve call sites are a little further from where a failure would normally be
contained. ClusterMessageSerializer defers GossipEnvelope's decompression into a
thunk, so it runs when the gossip is read rather than on a deserialization thread; and
Welcome is deserialized while the node is still uninitialized.

The Jackson serializers already bound this — JacksonSerializer.gunzip, added in #3491 —
but nothing else does.

Modification

Add org.apache.pekko.serialization.Decompression (@InternalApi), whose gunzip
stops as soon as the decompressed size passes pekko.serialization.max-decompressed-size
and reports it as a NotSerializableException. Route all twelve call sites through it:

Module	Helper	Sites
cluster	ClusterMessageSerializer.decompress	2 — Welcome, GossipEnvelope
distributed-data	SerializationSupport.decompress	6 — Gossip, ORSet, ORMap, LWWMap, PNCounterMap, ORMultiMap
cluster-tools	DistributedPubSubMessageSerializer.decompress	2 — Status, Delta
cluster-sharding	ClusterShardingMessageSerializer.decompress	1 — CoordinatorState
cluster-metrics	MessageSerializer.decompress	1 — MetricsGossipEnvelope

The setting defaults to unlimited, meaning no limit, so the bound is opt-in: a patch
release must not start rejecting payloads an existing cluster legitimately exchanges, and
there is no default that is provably above every deployment's largest ddata or gossip
payload. A negative number such as -1 is also accepted, matching the
pekko.serialization.jackson.read.max-document-length convention (dual spelling per
review on #3515, kept consistent here). Operators who want the protection set a size such
as 256 MiB, larger than anything their cluster legitimately sends. An unlimited maximum
skips the size check in gunzip; config's getBytes refuses both spellings, so the
setting is read as a string first and as a memory size only when it is neither
unlimited nor a negative number.

The four class-based serializers read the maximum once into a val.
SerializationSupport is a public trait whose implementors take system as a
constructor parameter, so it reads the maximum per call instead — a field there would
add abstract accessors to the trait and break binary compatibility, and the lookup is
negligible next to the decompression it guards.

The Jackson serializers are unchanged: they are already bounded and keep their own
pekko.serialization.jackson.compression.max-decompressed-size.

decompress signatures are unchanged, so there is no binary-compatibility impact.

Result

By default, behaviour is unchanged. With a maximum configured, a payload that expands
beyond it is rejected as an ordinary serialization failure naming the setting.

Tests

• sbt "actor-tests/testOnly org.apache.pekko.serialization.DecompressionSpec" — 9 passed (round trip, the limit boundary, one byte over, the setting named in the message, a highly compressible payload rejected without inflating, no limit applied for a negative maximum, the unlimited default, a configured size read correctly, and a configured -1 read as unlimited)
• sbt "cluster/testOnly org.apache.pekko.cluster.protobuf.*" — 13 passed, including a new ClusterMessageSerializerDecompressionSpec covering both the eager Welcome path and the deferred GossipEnvelope path
• sbt "distributed-data/testOnly org.apache.pekko.cluster.ddata.protobuf.*" — 30 passed, including a new SerializationSupportDecompressionSpec covering the per-call lookup in the trait
• sbt "cluster-tools/testOnly org.apache.pekko.cluster.pubsub.protobuf.*" — 1 passed
• sbt "cluster-sharding/testOnly org.apache.pekko.cluster.sharding.protobuf.*" — 14 passed
• sbt "cluster-metrics/testOnly org.apache.pekko.cluster.metrics.protobuf.*" — 9 passed
• sbt "actor/mimaReportBinaryIssues" "cluster/mimaReportBinaryIssues" "cluster-metrics/mimaReportBinaryIssues" "cluster-sharding/mimaReportBinaryIssues" "cluster-tools/mimaReportBinaryIssues" "distributed-data/mimaReportBinaryIssues" — no issues
• sbt scalafmtAll headerCreateAll — no changes

References

Extends #3491, which bounded decompression in the Jackson serializer only.

Decompression.scala consolidates the decompress bodies of the five Akka-derived
serializers listed above together with JacksonSerializer.gunzip, so it carries the
derived-from-Akka header and the Lightbend copyright. The three new specs are new code
and carry the standard ASF header.

[pjfanning]

I will submit a fix the merge conflict soon. The aim is to get this backported for the 1.7.1 release. The limit is opt-in to make it easier to roll-out.