This is an automated email from the ASF dual-hosted git repository.
He-Pin 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 07a346cc4b fix: validate full UniqueAddress in Artery system message
ACK/NACK handling (#3175)
07a346cc4b is described below
commit 07a346cc4bb7277cde2256426b6bb012664be4cf
Author: He-Pin(kerr) <[email protected]>
AuthorDate: Thu Jun 25 04:34:45 2026 +0800
fix: validate full UniqueAddress in Artery system message ACK/NACK handling
(#3175)
Motivation:
SystemMessageDelivery.notify() only compared the Address portion of
incoming ACK/NACK messages, ignoring the UID. When a remote node
restarts and obtains a new UID, a delayed ACK from the previous
incarnation (same address, different UID) could pass the check and
incorrectly clear the unacknowledged buffer, causing system messages
to be silently dropped.
Modification:
Add isFromCurrentRemote() helper that checks the full UniqueAddress
(address + UID) against the association's known uniqueRemoteAddress.
Falls back to address-only matching when uniqueRemoteAddress is not
yet established (pre-handshake), preserving existing behavior.
Result:
Stale ACKs from previous remote incarnations are rejected, preventing
incorrect removal of system messages from the resend buffer.
Tests:
- sbt "remote / Test / testOnly
org.apache.pekko.remote.artery.SystemMessageDeliverySpec" — all 9 tests pass
References:
Fixes #3174, Refs #3160
---
.../remote/artery/SystemMessageDelivery.scala | 10 ++++-
.../remote/artery/SystemMessageDeliverySpec.scala | 44 ++++++++++++++++++++++
2 files changed, 52 insertions(+), 2 deletions(-)
diff --git
a/remote/src/main/scala/org/apache/pekko/remote/artery/SystemMessageDelivery.scala
b/remote/src/main/scala/org/apache/pekko/remote/artery/SystemMessageDelivery.scala
index e9a59b743d..f1cdf18559 100644
---
a/remote/src/main/scala/org/apache/pekko/remote/artery/SystemMessageDelivery.scala
+++
b/remote/src/main/scala/org/apache/pekko/remote/artery/SystemMessageDelivery.scala
@@ -111,6 +111,12 @@ import pekko.util.PrettyDuration.PrettyPrintableDuration
private def remoteAddressLogParam: String =
outboundContext.associationState.uniqueRemoteAddress().getOrElse(remoteAddress).toString
+ private def isFromCurrentRemote(from: UniqueAddress): Boolean =
+ outboundContext.associationState.uniqueRemoteAddress() match {
+ case Some(uniqueRemote) => uniqueRemote == from
+ case None => from.address == remoteAddress
+ }
+
override protected def logSource: Class[?] =
classOf[SystemMessageDelivery]
override def preStart(): Unit = {
@@ -156,8 +162,8 @@ import pekko.util.PrettyDuration.PrettyPrintableDuration
// ControlMessageObserver, external call
override def notify(inboundEnvelope: InboundEnvelope): Unit = {
inboundEnvelope.message match {
- case ack: Ack => if (ack.from.address == remoteAddress)
ackCallback.invoke(ack)
- case nack: Nack => if (nack.from.address == remoteAddress)
nackCallback.invoke(nack)
+ case ack: Ack => if (isFromCurrentRemote(ack.from))
ackCallback.invoke(ack)
+ case nack: Nack => if (isFromCurrentRemote(nack.from))
nackCallback.invoke(nack)
case _ => // not interested
}
}
diff --git
a/remote/src/test/scala/org/apache/pekko/remote/artery/SystemMessageDeliverySpec.scala
b/remote/src/test/scala/org/apache/pekko/remote/artery/SystemMessageDeliverySpec.scala
index 73dff9928d..ba87667451 100644
---
a/remote/src/test/scala/org/apache/pekko/remote/artery/SystemMessageDeliverySpec.scala
+++
b/remote/src/test/scala/org/apache/pekko/remote/artery/SystemMessageDeliverySpec.scala
@@ -287,6 +287,50 @@ class SystemMessageDeliverySpec extends
AbstractSystemMessageDeliverySpec(System
sink.expectComplete()
}
+ "be ignored when ACK is from stale remote incarnation" in {
+ val replyProbe = TestProbe()
+ val controlSubject = new TestControlMessageSubject
+ val inboundContextB = new ManualReplyInboundContext(replyProbe.ref,
addressB, controlSubject)
+ val inboundContextA = new TestInboundContext(addressB, controlSubject)
+ val outboundContextA =
+
inboundContextA.association(addressB.address).asInstanceOf[TestOutboundContext]
+
+ // Drop msg-3 BEFORE inbound() so SystemMessageAcker never generates a
real ACK for it.
+ // This prevents the real ACK(3) from overwriting ACK(2) in
ManualReplyInboundContext.
+ val sink = send(sendCount = 3, resendInterval = 1.second,
outboundContextA)
+ .via(drop(dropSeqNumbers = Vector(3L)))
+ .via(inbound(inboundContextB))
+ .map(_.message.asInstanceOf[TestSysMsg])
+ .runWith(TestSink[TestSysMsg]())
+
+ sink.request(100)
+ sink.expectNext(TestSysMsg("msg-1"))
+ sink.expectNext(TestSysMsg("msg-2"))
+
+ // Complete handshake so uniqueRemoteAddress is set to the current
addressB
+ Await.result(outboundContextA.completeHandshake(addressB), 3.seconds)
+
+ // Deliver valid ACKs for msg-1 and msg-2 (msg-3 was dropped, no ACK
generated for it)
+ replyProbe.expectMsg(Ack(1L, addressB))
+ inboundContextB.deliverLastReply()
+ replyProbe.expectMsg(Ack(2L, addressB))
+ inboundContextB.deliverLastReply()
+
+ // Inject a stale ACK from a previous incarnation (same address,
different UID)
+ val staleAddress = UniqueAddress(addressB.address, addressB.uid + 1000)
+ controlSubject.sendControl(
+ InboundEnvelope(OptionVal.None, Ack(3L, staleAddress), OptionVal.None,
staleAddress.uid, OptionVal.None))
+
+ // msg-3 should NOT have been cleared by the stale ACK,
+ // so it must be resent after the resend interval
+ sink.expectNext(3.seconds, TestSysMsg("msg-3"))
+
+ // Deliver the real ACK from the resend to complete the stream
+ replyProbe.expectMsg(3.seconds, Ack(3L, addressB))
+ inboundContextB.deliverLastReply()
+ sink.expectComplete()
+ }
+
"deliver all during stress and random dropping" in {
val N = 500
val dropRate = 0.05
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
