From d0455c8096d87e40d6ca3c5a30a9a3a4d54c821d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=99=8E=E9=B8=A3?= Date: Thu, 25 Jun 2026 03:07:26 +0800 Subject: [PATCH] fix: validate full UniqueAddress in Artery system message ACK/NACK handling MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 ++++- .../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 e9a59b743d7..f1cdf185590 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 73dff9928d6..ba87667451b 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