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.git
The following commit(s) were added to refs/heads/main by this push:
new 4efa8195a5 test: cover the per-stream frame-size bound in TcpFraming
(#3524)
4efa8195a5 is described below
commit 4efa8195a59f61969a45c20a27e8993c8d6d9939
Author: PJ Fanning <[email protected]>
AuthorDate: Thu Sep 3 18:40:21 2026 +0100
test: cover the per-stream frame-size bound in TcpFraming (#3524)
Motivation:
TcpFraming.ReadStreamId selects maximumLargeFrameSize only for
ArteryTransport.LargeStreamId and maximumFrameSize otherwise (#3492),
but no test exercised that selection: the existing bound tests all use
a single maximumFrameSize with no maximumLargeFrameSize configured, so
they cannot tell the per-stream selection apart from a bug that applied
one bound to every stream.
Modification:
Add a perStreamBoundedFramingFlow fixture with distinct maximumFrameSize
and maximumLargeFrameSize, and two tests: a frame over the ordinary
maximum but within the large maximum is accepted on
ArteryTransport.LargeStreamId and rejected on
ArteryTransport.OrdinaryStreamId. Both frames carry their full declared
payload, so a false accept cannot hide behind truncation.
Result:
The per-stream bound selection in ReadStreamId is covered directly.
Tests:
- sbt "remote/testOnly org.apache.pekko.remote.artery.tcp.TcpFramingSpec" -
16 passed
- Checked the new tests discriminate: with ReadStreamId's selection
replaced by an unconditional maximumLargeFrameSize, the ordinary-stream
test fails ("Future.failed not completed with a throwable") while the
large-stream test still passes; reverted after confirming
- sbt "remote/scalafmtCheckAll" - clean
References:
None - test-coverage gap noticed while comparing TcpFraming.scala
(hardened in #3492) against an unrelated Akka fix for the same class of
issue
---
.../pekko/remote/artery/tcp/TcpFramingSpec.scala | 28 ++++++++++++++++++++++
1 file changed, 28 insertions(+)
diff --git
a/remote/src/test/scala/org/apache/pekko/remote/artery/tcp/TcpFramingSpec.scala
b/remote/src/test/scala/org/apache/pekko/remote/artery/tcp/TcpFramingSpec.scala
index 795f50876d..d7b9414991 100644
---
a/remote/src/test/scala/org/apache/pekko/remote/artery/tcp/TcpFramingSpec.scala
+++
b/remote/src/test/scala/org/apache/pekko/remote/artery/tcp/TcpFramingSpec.scala
@@ -36,6 +36,12 @@ class TcpFramingSpec extends PekkoSpec("""
private val maxFrameSize = 256 * 1024
private val boundedFramingFlow =
Flow[ByteString].via(new TcpFraming(acceptedMagic, maximumFrameSize =
maxFrameSize))
+ private val maxLargeFrameSize = maxFrameSize * 2
+ private val perStreamBoundedFramingFlow =
+ Flow[ByteString].via(new TcpFraming(
+ acceptedMagic,
+ maximumFrameSize = maxFrameSize,
+ maximumLargeFrameSize = maxLargeFrameSize))
private val payload5 = ByteString((1 to 5).map(_.toByte).toArray)
@@ -123,6 +129,28 @@ class TcpFramingSpec extends PekkoSpec("""
frames.head.byteBuffer.limit() should ===(maxFrameSize)
}
+ "accept, on the large-message stream, a frame over the ordinary maximum
but within the large maximum" in {
+ val length = maxFrameSize + 1
+ val payload = ByteString(Array.fill(length)(7.toByte))
+ val bytes = TcpFraming.encodeConnectionHeader(magic,
ArteryTransport.LargeStreamId) ++
+ encodeFrameHeader(length) ++ payload
+ val frames =
Source(List(bytes)).via(perStreamBoundedFramingFlow).runWith(Sink.seq).futureValue
+ frames.head.byteBuffer.limit() should ===(length)
+ frames.head.streamId should ===(ArteryTransport.LargeStreamId)
+ }
+
+ "reject, on the ordinary stream, a frame over the ordinary maximum even
though it is within the large maximum" in {
+ // the payload is included, at full length, so this can only fail via
the max-frame-size check -
+ // not via truncation - and so genuinely proves the ordinary stream is
bounded by maximumFrameSize,
+ // not the larger maximumLargeFrameSize that only applies to
ArteryTransport.LargeStreamId
+ val length = maxFrameSize + 1
+ val payload = ByteString(Array.fill(length)(7.toByte))
+ val bytes = TcpFraming.encodeConnectionHeader(magic,
ArteryTransport.OrdinaryStreamId) ++
+ encodeFrameHeader(length) ++ payload
+ val fail =
Source(List(bytes)).via(perStreamBoundedFramingFlow).runWith(Sink.seq).failed.futureValue
+ fail shouldBe a[FramingException]
+ }
+
"report truncated frames" in {
val bytes = TcpFraming.encodeConnectionHeader(magic, 3) ++
frameBytes(3).drop(1)
Source(List(bytes)).via(framingFlow).runWith(Sink.seq).failed.futureValue
shouldBe a[FramingException]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
