fix: deliver stashed timer messages for UnboundedStash and UnrestrictedStash (#3258)

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")
+    }
   }
 
 }

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)