Message-resend guard checks the sender instead of the destination, so it never suppresses resends

[aglinxinyuan]

What happened?

The 30-second message-resend loop's suppression guard tests the sender of a channel instead of the destination, so it never suppresses anything.

checkResend (PekkoMessageTransferService.scala:167-181):

if (refService.hasActorRef(channel.fromWorkerId)) {   // :175  <- the SENDER
  msgsNeedResend.foreach { msg => refService.forwardToActor(msg) }
}

channel here is an outgoing channel keyed ChannelIdentity(actorId, receiverId, ...), so fromWorkerId is this actor's own id — always registered (actorRefMapping(SELF) = actorService.self, PekkoActorRefMappingService.scala:48). The guard is therefore vacuously true. Delivery, meanwhile, routes by the destination: forwardToActor uses msg.internalMessage.channelId.toWorkerId (PekkoActorRefMappingService.scala:66).

Effect: unacked messages addressed to a stopped/removed worker are re-forwarded every 30 s forever. For a missing destination each resend lands in messageStash (:69-73) — unbounded growth plus a "unknown identifier" WARN per resend (:122) — or generates dead letters. This is also one of the amplifiers in the region-termination non-delivery hang (filed separately).

The fix looks one-line: test channel.toWorkerId — but note that today's vacuous guard is accidentally masking the question of what should happen to timed-out in-transit messages for a legitimately removed worker (drop? stash bounded? log once?), which deserves a decision in the same PR.

How to reproduce?

Code inspection is sufficient (the guard can never be false for an outgoing channel). Observable symptom: after any worker is stopped with unacked in-transit messages, messageStash for that worker grows by the full resend set every 30 s, with repeated unknown identifier WARNs.

Version/Branch

main (observed at 429be11; discovered during the investigation for #6916).