This is an automated email from the ASF dual-hosted git repository.
He-Pin 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 b2ae2fafa7 fix: deliver stashed timer messages for UnboundedStash and
UnrestrictedStash (#3264)
b2ae2fafa7 is described below
commit b2ae2fafa71b0f1f604d21d21f9009f62d7f2819
Author: He-Pin(kerr) <[email protected]>
AuthorDate: Sun Jul 5 16:59:14 2026 +0800
fix: deliver stashed timer messages for UnboundedStash and
UnrestrictedStash (#3264)
Motivation:
Timers rewrites currentMessage to the unwrapped timer message before
invoking
the user's receive, so that stash() captures the actual message rather than
the
internal TimerMsg wrapper. This rewrite was guarded by isInstanceOf[Stash],
which is false for actors mixing in UnboundedStash or UnrestrictedStash
directly
- they are siblings of Stash (all extend UnrestrictedStash), not subtypes.
As a
result stash() captured the TimerMsg wrapper, and on unstashAll() the
wrapper was
intercepted again and discarded (single-shot timer already removed, or stale
generation for a periodic timer), losing the message.
Modification:
- Guard the currentMessage rewrite with isInstanceOf[StashSupport] instead
of
isInstanceOf[Stash], so all three public stash traits (Stash,
UnboundedStash,
UnrestrictedStash - which all extend StashSupport) receive the unwrapped
message.
Result:
Timer messages stashed with UnboundedStash or UnrestrictedStash are no
longer
lost on unstash; behavior for Stash is unchanged and non-stashing actors are
unaffected.
Tests:
- actor-tests/testOnly org.apache.pekko.actor.TimerSpec
org.apache.pekko.actor.TimersAndStashSpec - all passed. Added directional
tests for UnboundedStash and UnrestrictedStash that fail before the fix
(the
timer message is lost on unstash).
- actor/mimaReportBinaryIssues - no binary issues.
References:
Fixes #3258
---
.../scala/org/apache/pekko/actor/TimerSpec.scala | 56 ++++++++++++++++++++++
.../main/scala/org/apache/pekko/actor/Timers.scala | 8 +++-
2 files changed, 62 insertions(+), 2 deletions(-)
diff --git a/actor-tests/src/test/scala/org/apache/pekko/actor/TimerSpec.scala
b/actor-tests/src/test/scala/org/apache/pekko/actor/TimerSpec.scala
index 2d51cf27ee..85e67b77ec 100644
--- a/actor-tests/src/test/scala/org/apache/pekko/actor/TimerSpec.scala
+++ b/actor-tests/src/test/scala/org/apache/pekko/actor/TimerSpec.scala
@@ -341,6 +341,44 @@ class TimersAndStashSpec extends PekkoSpec {
}
}
+ // Same scenario as ActorWithTimerAndStash but mixing in UnboundedStash,
which is a sibling of
+ // Stash (both extend UnrestrictedStash), to cover the timer/stash
interaction for it too (#3258).
+ class ActorWithTimerAndUnboundedStash(probe: ActorRef) extends Actor with
Timers with UnboundedStash {
+ timers.startSingleTimer("key", "scheduled", 50.millis)
+ def receive: Receive = stashing
+ def notStashing: Receive = {
+ case msg => probe ! msg
+ }
+
+ def stashing: Receive = {
+ case StopStashing =>
+ context.become(notStashing)
+ unstashAll()
+ case "scheduled" =>
+ probe ! "saw-scheduled"
+ stash()
+ }
+ }
+
+ // Same scenario mixing in UnrestrictedStash directly (needs an explicitly
configured
+ // deque-based mailbox, as it does not declare a RequiresMessageQueue)
(#3258).
+ class ActorWithTimerAndUnrestrictedStash(probe: ActorRef) extends Actor with
Timers with UnrestrictedStash {
+ timers.startSingleTimer("key", "scheduled", 50.millis)
+ def receive: Receive = stashing
+ def notStashing: Receive = {
+ case msg => probe ! msg
+ }
+
+ def stashing: Receive = {
+ case StopStashing =>
+ context.become(notStashing)
+ unstashAll()
+ case "scheduled" =>
+ probe ! "saw-scheduled"
+ stash()
+ }
+ }
+
"Timers combined with stashing" should {
"work" in {
@@ -350,6 +388,24 @@ class TimersAndStashSpec extends PekkoSpec {
actor ! StopStashing
probe.expectMsg("scheduled")
}
+
+ "work with UnboundedStash (#3258)" in {
+ val probe = TestProbe()
+ val actor = system.actorOf(Props(new
ActorWithTimerAndUnboundedStash(probe.ref)))
+ probe.expectMsg("saw-scheduled")
+ actor ! StopStashing
+ probe.expectMsg("scheduled")
+ }
+
+ "work with UnrestrictedStash (#3258)" in {
+ val probe = TestProbe()
+ val actor = system.actorOf(
+ Props(new ActorWithTimerAndUnrestrictedStash(probe.ref))
+ .withMailbox("pekko.actor.mailbox.unbounded-deque-based"))
+ probe.expectMsg("saw-scheduled")
+ actor ! StopStashing
+ probe.expectMsg("scheduled")
+ }
}
}
diff --git a/actor/src/main/scala/org/apache/pekko/actor/Timers.scala
b/actor/src/main/scala/org/apache/pekko/actor/Timers.scala
index 5b264ba4ac..9c57c6100b 100644
--- a/actor/src/main/scala/org/apache/pekko/actor/Timers.scala
+++ b/actor/src/main/scala/org/apache/pekko/actor/Timers.scala
@@ -55,8 +55,12 @@ trait Timers extends Actor {
case OptionVal.Some(m: AutoReceivedMessage) =>
context.asInstanceOf[ActorCell].autoReceiveMessage(Envelope(m,
self, context.system))
case OptionVal.Some(m) =>
- if (this.isInstanceOf[Stash]) {
- // this is important for stash interaction, as stash will look
directly at currentMessage #24557
+ if (this.isInstanceOf[StashSupport]) {
+ // This is important for stash interaction, as stash reads the
message directly from
+ // currentMessage (StashSupport) #24557. We match StashSupport
rather than Stash so that
+ // actors mixing in UnboundedStash or UnrestrictedStash directly
- which are siblings of
+ // Stash, not subtypes - also rewrite the unwrapped timer
message; otherwise stash()
+ // would re-stash the TimerMsg wrapper and the message would be
lost on unstash (#3258).
actorCell.currentMessage = actorCell.currentMessage.copy(message
= m)
}
super.aroundReceive(receive, m)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
