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 41dc602371 Tell a peer with a colliding Artery uid that it is
quarantined (#3676)
41dc602371 is described below
commit 41dc60237102ac87190d822d4a1368d0983f2eef
Author: PJ Fanning <[email protected]>
AuthorDate: Sat Oct 24 02:05:42 2026 +0100
Tell a peer with a colliding Artery uid that it is quarantined (#3676)
* Tell a peer with a colliding Artery uid that it is quarantined
Motivation:
When two Artery systems share a uid, the system that detects the
collision rejects the second peer's handshake, but the colliding peer
completes its own handshake from the receiver's HandshakeReq and its
messages still reach local actors. The receiver cannot drop them: the
envelope header only carries the origin uid, which resolves to the
existing association. The quarantine call on collision was a no-op
because the handshake had not completed.
Modification:
On UidCollisionException, ArteryTransport.completeHandshake sends the
existing Quarantined(localAddress, peer) control message to the
colliding peer instead of the no-op quarantine. Control messages are
injected after OutboundHandshake, so they are sent before the outbound
handshake completes.
Result:
The colliding peer logs that it was quarantined and publishes
ThisActorSystemQuarantinedEvent, which the Split Brain Resolver uses to
down the node. No wire format change; older Artery peers handle the
message the same way.
Tests:
- sbt "remote/testOnly
org.apache.pekko.remote.artery.RemoteConnectionSpec": collision tests fail
without the change, pass with it
References:
Refs #3672
* Unsubscribe event stream probes in RemoteConnectionSpec
Motivation:
Review: the warning and quarantined probes stayed subscribed to the
event stream after their assertions.
Modification:
Unsubscribe each probe once its expected event has been seen.
Result:
Tests leave the event stream subscriptions clean.
Tests:
- sbt "remote/testOnly
org.apache.pekko.remote.artery.RemoteConnectionSpec": 4 passed
References:
Refs #3672
* Log the Quarantined sent to a colliding peer and check the healthy peer
is not quarantined
---
.../pekko/remote/artery/ArteryTransport.scala | 9 ++++--
.../pekko/remote/artery/RemoteConnectionSpec.scala | 35 +++++++++++++++++++++-
2 files changed, 41 insertions(+), 3 deletions(-)
diff --git
a/remote/src/main/scala/org/apache/pekko/remote/artery/ArteryTransport.scala
b/remote/src/main/scala/org/apache/pekko/remote/artery/ArteryTransport.scala
index cb4df8adf9..15890226cf 100644
--- a/remote/src/main/scala/org/apache/pekko/remote/artery/ArteryTransport.scala
+++ b/remote/src/main/scala/org/apache/pekko/remote/artery/ArteryTransport.scala
@@ -788,8 +788,13 @@ private[remote] abstract class ArteryTransport(_system:
ExtendedActorSystem, _pr
case ShuttingDown => Future.successful(Done) // silence it
case exc: UidCollisionException =>
log.warning(exc.getMessage)
- // quarantine will not do much since handshake not completed, but for
good measures
- quarantine(peer.address, Some(peer.uid), exc.getMessage, harmless =
false)
+ // Quarantining the peer here would be ignored since the handshake is
not completed, and the
+ // uid belongs to the existing association. The envelope header only
carries the origin uid, so
+ // messages from the colliding peer cannot be told apart from the
existing association.
+ // Tell the colliding peer that it has been quarantined instead.
Control messages are not held
+ // back by the outbound handshake.
+ log.info("Sending Quarantined to colliding peer [{}]", peer)
+ sendControl(peer.address, Quarantined(localAddress, peer))
Future.failed(exc)
}
}
diff --git
a/remote/src/test/scala/org/apache/pekko/remote/artery/RemoteConnectionSpec.scala
b/remote/src/test/scala/org/apache/pekko/remote/artery/RemoteConnectionSpec.scala
index aa46f16877..42d15487eb 100644
---
a/remote/src/test/scala/org/apache/pekko/remote/artery/RemoteConnectionSpec.scala
+++
b/remote/src/test/scala/org/apache/pekko/remote/artery/RemoteConnectionSpec.scala
@@ -41,10 +41,33 @@ class RemoteConnectionSpec extends
ArteryMultiNodeSpec("pekko.remote.retry-gate-
}
// handshake retries can log the collision more than once, so expect at
least one warning
- private def expectCollisionWarning(warnings: TestProbe): Unit =
+ private def expectCollisionWarning(warnings: TestProbe): Unit = {
warnings.fishForMessage(hint = "UID collision warning") {
case w: Logging.Warning => w.message.toString.startsWith("UID collision
old [")
}
+ warnings.system.eventStream.unsubscribe(warnings.ref)
+ }
+
+ private def quarantinedProbe(system: ActorSystem): TestProbe = {
+ val probe = new TestProbe(system)
+ system.eventStream.subscribe(probe.ref,
classOf[ThisActorSystemQuarantinedEvent])
+ probe
+ }
+
+ // the system that detects the collision tells the colliding system that it
has been quarantined
+ private def expectQuarantinedBy(quarantined: TestProbe, system:
ActorSystem): Unit = {
+ val quarantinedBy =
RARP(system).provider.transport.asInstanceOf[ArteryTransport].localAddress
+ quarantined.fishForMessage(hint = "ThisActorSystemQuarantinedEvent") {
+ case e: ThisActorSystemQuarantinedEvent => e.remoteAddress ==
quarantinedBy
+ }
+ quarantined.system.eventStream.unsubscribe(quarantined.ref)
+ }
+
+ // the healthy system shares the colliding uid, so make sure the Quarantined
did not reach it
+ private def expectNotQuarantined(quarantined: TestProbe): Unit = {
+ quarantined.expectNoMessage(500.millis)
+ quarantined.system.eventStream.unsubscribe(quarantined.ref)
+ }
"Remoting between systems" should {
@@ -62,6 +85,7 @@ class RemoteConnectionSpec extends
ArteryMultiNodeSpec("pekko.remote.retry-gate-
pekko.remote.artery.canonical.port=$remotePort1
"""))
val remote1Probe = new TestProbe(remoteSystem1)
+ val remote1Quarantined = quarantinedProbe(remoteSystem1)
val remotePort2 = freePort()
val remoteSystem2 =
@@ -71,6 +95,7 @@ class RemoteConnectionSpec extends
ArteryMultiNodeSpec("pekko.remote.retry-gate-
pekko.remote.artery.canonical.port=$remotePort2
"""))
val remote2Probe = new TestProbe(remoteSystem2)
+ val remote2Quarantined = quarantinedProbe(remoteSystem2)
localProbe.expectNoMessage(2.seconds)
localSystem.actorOf(TestActors.echoActorProps, echoName)
@@ -97,6 +122,7 @@ class RemoteConnectionSpec extends
ArteryMultiNodeSpec("pekko.remote.retry-gate-
// doesn't get through
localProbe.expectNoMessage()
expectCollisionWarning(collisionWarnings1)
+ expectQuarantinedBy(remote2Quarantined, localSystem)
RARP(localSystem).provider.transport
.asInstanceOf[ArteryTransport]
.association(RARP(remoteSystem2).provider.getDefaultAddress)
@@ -128,6 +154,8 @@ class RemoteConnectionSpec extends
ArteryMultiNodeSpec("pekko.remote.retry-gate-
// still works in other direction
selectionFromRemote1ToLocal.tell("ping1b again", remote1Probe.ref)
remote1Probe.expectMsg("ping1b again")
+
+ expectNotQuarantined(remote1Quarantined)
}
"handle uid collision when connection FROM two systems with same uid" in {
@@ -145,6 +173,7 @@ class RemoteConnectionSpec extends
ArteryMultiNodeSpec("pekko.remote.retry-gate-
pekko.remote.artery.canonical.port=$remotePort1
"""))
val remote1Probe = new TestProbe(remoteSystem1)
+ val remote1Quarantined = quarantinedProbe(remoteSystem1)
val remotePort2 = freePort()
val remoteSystem2 =
@@ -154,6 +183,7 @@ class RemoteConnectionSpec extends
ArteryMultiNodeSpec("pekko.remote.retry-gate-
pekko.remote.artery.canonical.port=$remotePort2
"""))
val remote2Probe = new TestProbe(remoteSystem2)
+ val remote2Quarantined = quarantinedProbe(remoteSystem2)
localProbe.expectNoMessage(2.seconds)
localSystem.actorOf(TestActors.echoActorProps, echoName)
@@ -180,6 +210,7 @@ class RemoteConnectionSpec extends
ArteryMultiNodeSpec("pekko.remote.retry-gate-
// doesn't get through
localProbe.expectNoMessage()
expectCollisionWarning(collisionWarnings3)
+ expectQuarantinedBy(remote2Quarantined, localSystem)
RARP(localSystem).provider.transport
.asInstanceOf[ArteryTransport]
.association(RARP(remoteSystem2).provider.getDefaultAddress)
@@ -211,6 +242,8 @@ class RemoteConnectionSpec extends
ArteryMultiNodeSpec("pekko.remote.retry-gate-
// still works in other direction
selectionFromLocalToRemote1.tell("ping1a again", localProbe.ref)
localProbe.expectMsg("ping1a again")
+
+ expectNotQuarantined(remote1Quarantined)
}
"be able to connect to system even if it's not there at first" in {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
