This is an automated email from the ASF dual-hosted git repository.
pjfanning pushed a commit to branch 1.7.x
in repository https://gitbox.apache.org/repos/asf/pekko.git
The following commit(s) were added to refs/heads/1.7.x by this push:
new f2b7bdbf08 fix: bound the number of entries in a compression table
advertisement (#3510) (#3521)
f2b7bdbf08 is described below
commit f2b7bdbf0813ccf087f432cd8470e2b5a21e072b
Author: PJ Fanning <[email protected]>
AuthorDate: Fri Sep 4 14:41:24 2026 +0100
fix: bound the number of entries in a compression table advertisement
(#3510) (#3521)
Motivation:
deserializeCompressionAdvertisement resolves every key in the advertised
table, and for actor refs that means parsing a path and populating the
resolve cache. The key list was unbounded, so the only limit was the
transport frame size. Measured, 10000 entries is 369 KB of wire and about
150 ms of CPU on the inbound control stream, against 9 KB for the 256
entry table a peer legitimately advertises.
Modification:
Reject an advertisement carrying more entries than
pekko.remote.artery.advanced.compression.<table>.max, the setting that
bounds the table on the sending side and is normally the same across a
cluster. When it is "off" locally there is no number to check against and
no bound is applied.
Result:
An oversized advertisement is reported as a serialization failure, which
the inbound stream logs and drops. Advertisements are resent periodically,
so a dropped one costs at most a delay in establishing compression.
---
.../serialization/ArteryMessageSerializer.scala | 35 ++++++++++++++++++++--
.../ArteryMessageSerializerSpec.scala | 26 +++++++++++++++-
2 files changed, 58 insertions(+), 3 deletions(-)
diff --git
a/remote/src/main/scala/org/apache/pekko/remote/serialization/ArteryMessageSerializer.scala
b/remote/src/main/scala/org/apache/pekko/remote/serialization/ArteryMessageSerializer.scala
index 5fe8b53bd1..73c1ccf37d 100644
---
a/remote/src/main/scala/org/apache/pekko/remote/serialization/ArteryMessageSerializer.scala
+++
b/remote/src/main/scala/org/apache/pekko/remote/serialization/ArteryMessageSerializer.scala
@@ -27,6 +27,7 @@ import pekko.remote.artery.compress.CompressionProtocol._
import pekko.serialization.{ BaseSerializer, Serialization,
SerializationExtension, SerializerWithStringManifest }
import pekko.remote.artery.Flush
import pekko.remote.artery.FlushAck
+import pekko.util.Helpers.toRootLowerCase
/** INTERNAL API */
private[pekko] object ArteryMessageSerializer {
@@ -60,6 +61,18 @@ private[pekko] final class ArteryMessageSerializer(val
system: ExtendedActorSyst
private lazy val serialization = SerializationExtension(system)
+ // `pekko.remote.artery.advanced.compression.<table>.max` bounds the number
of entries the
+ // sending side puts in a table, and is normally the same setting across a
cluster. Parsed the
+ // same way ArterySettings parses it, without building the whole settings
object here.
+ private def compressionMax(table: String): Int = {
+ val path = s"pekko.remote.artery.advanced.compression.$table.max"
+ if (toRootLowerCase(system.settings.config.getString(path)) == "off") 0
+ else system.settings.config.getInt(path)
+ }
+
+ private val maxActorRefCompressionEntries: Int = compressionMax("actor-refs")
+ private val maxClassManifestCompressionEntries: Int =
compressionMax("manifests")
+
override def manifest(o: AnyRef): String = o match { // most frequent ones
first
case _: SystemMessageDelivery.SystemMessageEnvelope =>
SystemMessageEnvelopeManifest
case _: SystemMessageDelivery.Ack =>
SystemMessageDeliveryAckManifest
@@ -123,7 +136,11 @@ private[pekko] final class ArteryMessageSerializer(val
system: ExtendedActorSyst
case ActorRefCompressionAdvertisementAckManifest =>
deserializeCompressionTableAdvertisementAck(bytes,
ActorRefCompressionAdvertisementAck.apply)
case ClassManifestCompressionAdvertisementManifest =>
- deserializeCompressionAdvertisement(bytes, identity,
ClassManifestCompressionAdvertisement.apply)
+ deserializeCompressionAdvertisement(
+ bytes,
+ identity,
+ maxClassManifestCompressionEntries,
+ ClassManifestCompressionAdvertisement.apply)
case ClassManifestCompressionAdvertisementAckManifest =>
deserializeCompressionTableAdvertisementAck(bytes,
ClassManifestCompressionAdvertisementAck.apply)
case ArteryHeartbeatManifest => RemoteWatcher.ArteryHeartbeat
@@ -158,7 +175,11 @@ private[pekko] final class ArteryMessageSerializer(val
system: ExtendedActorSyst
serializeCompressionAdvertisement(adv)(serializeActorRef)
def deserializeActorRefCompressionAdvertisement(bytes: Array[Byte]):
ActorRefCompressionAdvertisement =
- deserializeCompressionAdvertisement(bytes, deserializeActorRef,
ActorRefCompressionAdvertisement.apply)
+ deserializeCompressionAdvertisement(
+ bytes,
+ deserializeActorRef,
+ maxActorRefCompressionEntries,
+ ActorRefCompressionAdvertisement.apply)
def serializeCompressionAdvertisement[T](adv: CompressionAdvertisement[T])(
keySerializer: T => String):
ArteryControlFormats.CompressionTableAdvertisement = {
@@ -179,9 +200,19 @@ private[pekko] final class ArteryMessageSerializer(val
system: ExtendedActorSyst
def deserializeCompressionAdvertisement[T, U](
bytes: Array[Byte],
keyDeserializer: String => T,
+ maxEntries: Int,
create: (UniqueAddress, CompressionTable[T]) => U): U = {
val protoAdv =
ArteryControlFormats.CompressionTableAdvertisement.parseFrom(bytes)
+ // Every key is resolved, and for actor refs that means parsing a path and
populating the
+ // resolve cache, so a message with far more entries than a table can hold
is work out of
+ // proportion to its size. `maxEntries` is 0 when compression is switched
off here, and then
+ // there is no configured number to check against.
+ if (maxEntries > 0 && protoAdv.getKeysCount > maxEntries)
+ throw new NotSerializableException(
+ s"Compression table advertisement carries [${protoAdv.getKeysCount}]
entries, more than " +
+ s"the configured maximum of [$maxEntries]")
+
val kvs =
protoAdv.getKeysList.asScala
.map(keyDeserializer)
diff --git
a/remote/src/test/scala/org/apache/pekko/remote/serialization/ArteryMessageSerializerSpec.scala
b/remote/src/test/scala/org/apache/pekko/remote/serialization/ArteryMessageSerializerSpec.scala
index bc01ea2e78..88c6167881 100644
---
a/remote/src/test/scala/org/apache/pekko/remote/serialization/ArteryMessageSerializerSpec.scala
+++
b/remote/src/test/scala/org/apache/pekko/remote/serialization/ArteryMessageSerializerSpec.scala
@@ -19,7 +19,7 @@ import org.apache.pekko
import pekko.actor._
import pekko.remote.artery.Flush
import pekko.remote.artery.FlushAck
-import pekko.remote.{ RemoteWatcher, UniqueAddress }
+import pekko.remote.{ ArteryControlFormats, RemoteWatcher, UniqueAddress }
import pekko.remote.artery.{ ActorSystemTerminating,
ActorSystemTerminatingAck, Quarantined, SystemMessageDelivery }
import pekko.remote.artery.OutboundHandshake.{ HandshakeReq, HandshakeRsp }
import pekko.remote.artery.compress.CompressionProtocol.{
@@ -74,6 +74,30 @@ class ArteryMessageSerializerSpec extends PekkoSpec {
"not support UniqueAddresses without host/port set" in pending
+ "reject a compression table advertisement with more entries than the
configured maximum" in {
+ val serializer = new
ArteryMessageSerializer(system.asInstanceOf[ExtendedActorSystem])
+ val max =
system.settings.config.getInt("pekko.remote.artery.advanced.compression.actor-refs.max")
+
+ def advertisement(entries: Int): Array[Byte] = {
+ val builder =
ArteryControlFormats.CompressionTableAdvertisement.newBuilder
+ .setFrom(serializer.serializeUniqueAddress(uniqueAddress()))
+ .setOriginUid(17L)
+ .setTableVersion(1)
+ (0 until entries).foreach { i =>
+ builder.addKeys(s"pekko://sys@host:1234/user/a$i")
+ builder.addValues(i)
+ }
+ builder.build().toByteArray
+ }
+
+ // a table of exactly the configured size is what a peer legitimately
advertises
+ serializer.fromBinary(advertisement(max), "f") shouldBe
a[ActorRefCompressionAdvertisement]
+
+ intercept[NotSerializableException] {
+ serializer.fromBinary(advertisement(max + 1), "f")
+ }.getMessage should include(s"more than the configured maximum of
[$max]")
+ }
+
"reject invalid manifest" in {
intercept[IllegalArgumentException] {
val serializer = new
ArteryMessageSerializer(system.asInstanceOf[ExtendedActorSystem])
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
