pjfanning commented on code in PR #3702:
URL: https://github.com/apache/pekko/pull/3702#discussion_r4371818744
##########
remote/src/main/scala/org/apache/pekko/remote/artery/Handshake.scala:
##########
@@ -104,6 +104,17 @@ private[remote] class OutboundHandshake(
}
override def postStop(): Unit = {
+ pendingMessage match {
+ case OptionVal.Some(envelope) =>
+ // the envelope has already been grabbed from the upstream
SendQueue, so
+ // apply the same safeguards as SendQueue.postStop instead of
dropping it silently
+ pendingMessage = OptionVal.None
+ system.deadLetters ! envelope
+ if (envelope.message.isInstanceOf[SystemMessage])
+ outboundContext.quarantine(
+ s"OutboundHandshake stopped with pending system message
[${envelope.message.getClass.getName}].")
Review Comment:
`Association.quarantine` sends `ClearSystemMessageDelivery` and
`sendControl(Quarantined(...))`. Called from this stage's `postStop`, those
messages go to the old control `SendQueue`, which is being torn down and may
already have drained in its own `postStop`. `lazyRestart` then creates a fresh
queue, so the peer may never learn it was quarantined and
`SystemMessageDelivery` may not be cleared for that incarnation.
##########
remote/src/main/scala/org/apache/pekko/remote/artery/Handshake.scala:
##########
@@ -104,6 +104,17 @@ private[remote] class OutboundHandshake(
}
override def postStop(): Unit = {
+ pendingMessage match {
+ case OptionVal.Some(envelope) =>
+ // the envelope has already been grabbed from the upstream
SendQueue, so
+ // apply the same safeguards as SendQueue.postStop instead of
dropping it silently
+ pendingMessage = OptionVal.None
+ system.deadLetters ! envelope
+ if (envelope.message.isInstanceOf[SystemMessage])
+ outboundContext.quarantine(
Review Comment:
This quarantine also fires when the stream stops normally, for example when
the transport or ActorSystem shuts down via the killSwitch while a system
message (a Watch, say) is held behind an injected `HandshakeReq`. That produces
a WARN saying the connection irrecoverably failed and a `QuarantinedEvent`
instead of a `GracefulShutdownQuarantinedEvent`, and it tries to
`sendControl(Quarantined)` to the peer during an orderly shutdown. Failures of
the control stream are already quarantined in `attachOutboundStreamRestart`,
and stops triggered by a quarantine are already quarantined, so should this
skip the quarantine (or use `harmless = true`) when the transport or
association is shutting down?
##########
remote/src/main/scala/org/apache/pekko/remote/artery/Handshake.scala:
##########
@@ -104,6 +104,17 @@ private[remote] class OutboundHandshake(
}
override def postStop(): Unit = {
+ pendingMessage match {
+ case OptionVal.Some(envelope) =>
+ // the envelope has already been grabbed from the upstream
SendQueue, so
+ // apply the same safeguards as SendQueue.postStop instead of
dropping it silently
+ pendingMessage = OptionVal.None
+ system.deadLetters ! envelope
+ if (envelope.message.isInstanceOf[SystemMessage])
+ outboundContext.quarantine(
+ s"OutboundHandshake stopped with pending system message
[${envelope.message.getClass.getName}].")
+ case _ =>
+ }
outboundContext.associationState.removeUniqueRemoteAddressListener(uniqueRemoteAddressListener)
Review Comment:
`removeUniqueRemoteAddressListener` and `super.postStop()` now run after the
new dead-letter and quarantine code, with no `try/finally`. If `quarantine()`
or the dead-letter send throws (for example during a racing shutdown), the
listener stays registered on `AssociationState` and keeps the dead stage logic
alive for the life of the association. Could the cleanup go in a `finally`?
##########
remote/src/test/scala/org/apache/pekko/remote/artery/OutboundHandshakeSpec.scala:
##########
@@ -165,5 +171,75 @@ class OutboundHandshakeSpec extends PekkoSpec("""
}
}
+}
+
+class OutboundHandshakePendingMessageSpec extends OutboundHandshakeSpecBase("""
Review Comment:
These tests only check a local `deadLetters` and
`TestOutboundContext.quarantine`, which sets the quarantined state
unconditionally. The real `Association.quarantine` does nothing when the UID is
unknown or doesn't match, and it sends control messages, so these tests would
still pass if production quarantine did nothing or lost those messages. A test
against a real `Association` (or a multi-node or ArteryTransport-level test)
would cover the production path.
##########
remote/src/test/scala/org/apache/pekko/remote/artery/OutboundHandshakeSpec.scala:
##########
@@ -165,5 +171,75 @@ class OutboundHandshakeSpec extends PekkoSpec("""
}
}
+}
+
+class OutboundHandshakePendingMessageSpec extends OutboundHandshakeSpecBase("""
+ pekko.scheduler.implementation =
"org.apache.pekko.testkit.ExplicitlyTriggeredScheduler"
+ pekko.stream.materializer.dispatcher =
"pekko.test.calling-thread-dispatcher"
+ """) {
+
+ private def advanceHandshakeTimer(): Unit =
+
system.scheduler.asInstanceOf[ExplicitlyTriggeredScheduler].timePasses(1.second)
+
+ "OutboundHandshake stage" must {
+ "send pending message to deadLetters when stopped while holding it" in {
+ val inboundContext = new TestInboundContext(localAddress = addressA)
+ val outboundContext = inboundContext.association(addressB.address)
+ inboundContext.completeHandshake(addressB)
+ val (upstream, downstream) = setupStream(outboundContext,
injectHandshakeInterval = 1.second)
+
+ system.eventStream.subscribe(testActor, classOf[DeadLetter])
+
+ downstream.request(1)
+ downstream.expectNext(HandshakeReq(addressA, addressB.address))
+ // The calling-thread stream dispatcher handles the scheduled callback
+ // before timePasses returns, so the next message is held behind a
handshake.
+ advanceHandshakeTimer()
+
+ downstream.request(1)
+ upstream.sendNext("msg1")
+ downstream.expectNext(HandshakeReq(addressA, addressB.address))
+
+ // the stage is now stopped while holding msg1 in pendingMessage
+ downstream.cancel()
+
+ fishForMessage(hint = "envelope with msg1 in deadLetters") {
+ case DeadLetter(env: OutboundEnvelope, _, _) => env.message == "msg1"
+ case _ => false
+ }
+ outboundContext.associationState.isQuarantined() should ===(false)
+ system.eventStream.unsubscribe(testActor, classOf[DeadLetter])
Review Comment:
The unsubscribe only runs if the test passes. If `fishForMessage` or the
`isQuarantined` assertion fails, `testActor` stays subscribed to `DeadLetter`
and the next test gets stray DeadLetters. Could the body go in `try/finally`?
The same applies to the second test.
##########
remote/src/main/scala/org/apache/pekko/remote/artery/Handshake.scala:
##########
@@ -104,6 +104,17 @@ private[remote] class OutboundHandshake(
}
override def postStop(): Unit = {
+ pendingMessage match {
Review Comment:
This repeats the policy in `Association.sendQueuePostStop`, and the copies
already behave differently: `SendQueue` quarantines only for the control
stream, but this quarantines for every lane (ordinary and large streams too),
and the reason strings and counting differ. Could `OutboundHandshake` take a
postStop callback, like `SendQueue`'s `postStopAction`, so there is one place
that decides between dead letters and quarantine?
More broadly, this covers only the one envelope held by this stage. System
messages already pulled into other buffered or async stages upstream of
`SystemMessageDelivery` can still be dropped silently when the stream stops. A
general check when the outbound control stream terminates might be a more
complete fix for #3690.
##########
docs/src/main/paradox/remoting-artery.md:
##########
@@ -344,6 +344,8 @@ partition heals. A cluster member is not quarantined when
the failure detector t
* Overflow of the system message delivery buffer, e.g. because of too many
`watch` requests at the same time
(`system-message-buffer-size` config).
* Unexpected exception occurs in the control subchannel of the remoting
infrastructure.
+ * An outbound stream stops with undelivered system messages, including a
message held while sending a periodic
Review Comment:
This promises more than the code does. The `SendQueue` of the ordinary and
large streams (`sendToDeadLetters`) never quarantines, and stops caused by
shutdown or quarantine are handled separately. Only the control stream and the
new handshake path quarantine, so this should probably be narrowed to the
control stream, or the code changed to match.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
