fix: cancel delayed restart when remembering entities and eagerly restarting (#32749)

Author: leviramsey

akka-cluster-sharding/src/main/scala/akka/cluster/sharding/Shard.scala

@@ -989,6 +989,7 @@ private[akka] class Shard(
             log.debug("{}: Passivation started for [{}]", typeName, id)
           entities.entityPassivating(id)
           entity ! stopMessage
+
           val passivationTimeout = PassivationTimedOut(entity)
           timers.startSingleTimer(
             passivationTimeout,
@@ -1144,7 +1145,12 @@ private[akka] class Shard(
   def getOrCreateEntity(id: EntityId): ActorRef = {
     entities.entity(id) match {
       case OptionVal.Some(child) => child
-      case _ =>
+      case _                     =>
+        // if we're remembering entities, there's a chance we have a timer to restart, which is now moot
+        if (rememberEntitiesStore.isDefined && entities.entityState(id) == WaitingForRestart) {
+          timers.cancel(RestartTerminatedEntity(id))
+        }
+
         val name = URLEncoder.encode(id, "utf-8")
         val a = context.actorOf(entityProps(id), name)
         context.watchWith(a, EntityTerminated(a))

akka-cluster-sharding/src/test/scala/akka/cluster/sharding/RememberEntitiesFailureSpec.scala

@@ -54,6 +54,9 @@ object RememberEntitiesFailureSpec {
         context.stop(self)
       case "graceful-stop" =>
         context.parent ! ShardRegion.Passivate("stop")
+      case "incarnation" =>
+        // Don't do this at home, kids...
+        sender() ! self
       case msg => sender() ! msg
     }
   }
@@ -395,6 +398,53 @@ class RememberEntitiesFailureSpec
         system.stop(sharding)
       }
     }
+
+    "neither restart entity nor crash when entity passivates after being eagerly restarted" in {
+      val shardingSettings = ClusterShardingSettings(system).withRememberEntities(true)
+
+      var sharding: ActorRef = null
+      val probe = TestProbe()
+      implicit val sender: ActorRef = probe.ref
+
+      val spawnProbe = TestProbe()
+      val props = Props[EntityActor] {
+        spawnProbe.ref.tell("spawned", ActorRef.noSender)
+        new EntityActor()
+      }
+
+      sharding = ClusterSharding(system).start(
+        "failStartPassivate",
+        props,
+        shardingSettings,
+        extractEntityId,
+        extractShardId,
+        ShardAllocationStrategy.leastShardAllocationStrategy(absoluteLimit = 1, relativeLimit = 0.1),
+        "graceful-stop")
+
+      sharding ! EntityEnvelope(1, "incarnation")
+      spawnProbe.expectMsg("spawned")
+      var currentIncarnation = probe.receiveN(1).head.asInstanceOf[ActorRef]
+
+      probe.watch(currentIncarnation)
+      currentIncarnation ! "stop"
+      probe.expectTerminated(currentIncarnation)
+      // The restart timer is active
+      currentIncarnation = null
+
+      // restart the entity early
+      sharding ! EntityEnvelope(1, "incarnation")
+      spawnProbe.expectMsg("spawned")
+      sharding ! EntityEnvelope(1, "graceful-stop")
+      currentIncarnation = probe.receiveN(1).head.asInstanceOf[ActorRef]
+
+      probe.watch(currentIncarnation)
+      probe.expectTerminated(currentIncarnation)
+      // entity is now passivated
+
+      probe.watch(sharding)
+      spawnProbe.expectNoMessage(1.second)
+      probe.expectNoMessage(1.second)
+    }
   }
 
 }