He-Pin opened a new issue, #3690:
URL: https://github.com/apache/pekko/issues/3690
## Summary
`OutboundHandshake` grabs an envelope from upstream and holds it in
`pendingMessage` while it cannot push downstream. Its `postStop` does not
account for that held envelope, so if the stage stops while holding one, the
envelope is neither pushed, nor returned to the send queue, nor reported — and
it bypasses the safeguard that quarantines an association stopped with pending
system messages.
## Evidence
`remote/src/main/scala/org/apache/pekko/remote/artery/Handshake.scala:82`:
```scala
private var pendingMessage: OptionVal[OutboundEnvelope] =
OptionVal.None
```
It is populated in `onPush` (`:112-131`) — at `:129` on the first push,
because `pushHandshakeReq()` has just consumed the downstream availability, and
at `:125` when `out` is not available:
```scala
if (injectHandshakeTickScheduled) {
// out is always available here, except for if a liveness
HandshakeReq was just pushed
if (isAvailable(out))
push(out, grab(in))
else {
if (pendingMessage.isDefined)
throw new IllegalStateException(s"pendingMessage expected to
be empty")
pendingMessage = OptionVal.Some(grab(in))
}
} else {
pushHandshakeReq()
pendingMessage = OptionVal.Some(grab(in))
}
```
It is only drained in `onPull` (`:133-143`):
```scala
case Completed =>
pendingMessage match {
case OptionVal.Some(p) =>
push(out, p)
pendingMessage = OptionVal.None
```
`postStop` (`:106-109`) does not touch it:
```scala
override def postStop(): Unit = {
outboundContext.associationState.removeUniqueRemoteAddressListener(uniqueRemoteAddressListener)
super.postStop()
}
```
By the time the envelope is in `pendingMessage` it has already been
`grab`bed from `in`, i.e. removed from the upstream `SendQueue`. The queue's
own stop handler therefore cannot see it —
`remote/src/main/scala/org/apache/pekko/remote/artery/Association.scala:793-798`:
```scala
def sendQueuePostStop[T](pending: Vector[OutboundEnvelope]): Unit = {
sendToDeadLetters(pending)
val systemMessagesCount = pending.count(env =>
env.message.isInstanceOf[SystemMessage])
if (systemMessagesCount > 0)
quarantine(s"SendQueue stopped with [$systemMessagesCount] pending
system messages.")
}
```
## Impact
An envelope lost this way skips both `sendToDeadLetters` and the
`systemMessagesCount > 0` quarantine check. If the lost envelope carried a
`SystemMessage` (`Watch`, `Unwatch`, `Terminate`), the association is torn down
without the quarantine that exists precisely to prevent silent system-message
loss, so the peer may keep believing the watch or termination is in flight.
The window is narrow — at most one envelope per stream, and only when the
stage stops while holding it — so this is unlikely to be observed often. It is
worth closing because the existing safeguard at `Association.scala:796-797`
shows the intent is that no pending system message should ever disappear
quietly.
## Suggested direction
In `postStop`, hand the held envelope back rather than dropping it: route it
through the same path `sendQueuePostStop` uses (dead letters plus the
pending-system-message accounting), or expose it so `Association` can include
it in that count.
## Test
Deterministically triggering this needs a way to stop the stage while
`pendingMessage` is non-empty — for example a stream test that withholds
downstream demand after the first element and then aborts the association,
asserting the envelope reaches dead letters (and that a `SystemMessage` payload
triggers quarantine).
--
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]
