diff --git a/build.sbt b/build.sbt index 6312c1f..ef1bf5a 100644 --- a/build.sbt +++ b/build.sbt @@ -68,6 +68,7 @@ lazy val catsActors = crossProject(JSPlatform, JVMPlatform, NativePlatform) libraryDependencies ++= Seq( "org.typelevel" %%% "cats-effect" % "3.7.0", "org.scalatest" %%% "scalatest" % "3.2.18" % Test, + "org.typelevel" %%% "cats-effect-testkit" % "3.7.0" % Test, "org.typelevel" %%% "cats-effect-testing-scalatest" % "1.8.0" % Test ) ) diff --git a/shared/src/main/scala/com/suprnation/actor/dungeon/ReceiveTimeout.scala b/shared/src/main/scala/com/suprnation/actor/dungeon/ReceiveTimeout.scala index 5a45a23..af96cd9 100644 --- a/shared/src/main/scala/com/suprnation/actor/dungeon/ReceiveTimeout.scala +++ b/shared/src/main/scala/com/suprnation/actor/dungeon/ReceiveTimeout.scala @@ -17,8 +17,8 @@ package com.suprnation.actor.dungeon import cats.Applicative -import cats.effect.{Ref, Sync} -import cats.syntax.flatMap._ +import cats.effect.{Clock, Ref, Sync} +import cats.syntax.all._ import com.suprnation.actor.dungeon.ReceiveTimeout.ReceiveTimeoutContext import scala.concurrent.duration.FiniteDuration @@ -35,42 +35,53 @@ class ReceiveTimeout[F[_]: Sync, Request]( receiveTimeoutContextRef: Ref[F, ReceiveTimeout.ReceiveTimeoutContext[Request]] ) { + // Use the cats-effect clock instead of System.currentTimeMillis so that receive-timeouts honour + // virtual time under cats.effect.testkit.TestControl (deterministic, instant tests). We use + // `monotonic` (not `realTime`): a timeout measures *elapsed* time, so it must be immune to + // wall-clock jumps (NTP, manual clock changes). TestControl advances monotonic with virtual time. + private def nowMillis: F[Long] = Clock[F].monotonic.map(_.toMillis) + def setReceiveTimeout(timeout: FiniteDuration, onTimeout: => Request): F[Unit] = - receiveTimeoutContextRef.set( - ReceiveTimeoutContext[Request]( - Some(timeout), - Some(System.currentTimeMillis()), - Some(onTimeout) + nowMillis.flatMap { now => + receiveTimeoutContextRef.set( + ReceiveTimeoutContext[Request]( + Some(timeout), + Some(now), + Some(onTimeout) + ) ) - ) + } def cancelReceiveTimeout: F[Unit] = receiveTimeoutContextRef.update(_.copy(receiveTimeout = None, message = None)) def markLastMessageTimestamp: F[Unit] = - receiveTimeoutContextRef.update { receiveTimeoutContext => - if (receiveTimeoutContext.receiveTimeout.isDefined) { - receiveTimeoutContext.copy(lastMessageTimestamp = Some(System.currentTimeMillis())) - } else { - receiveTimeoutContext + nowMillis.flatMap { now => + receiveTimeoutContextRef.update { receiveTimeoutContext => + if (receiveTimeoutContext.receiveTimeout.isDefined) { + receiveTimeoutContext.copy(lastMessageTimestamp = Some(now)) + } else { + receiveTimeoutContext + } } } def checkTimeout(action: Request => F[Any]): F[Any] = receiveTimeoutContextRef.get.flatMap { case ReceiveTimeoutContext(Some(timeout), Some(timestamp), Some(message)) => - val timeoutTime: Long = timestamp + timeout.toMillis - val currentTime: Long = System.currentTimeMillis() - if (timeoutTime <= currentTime) { - receiveTimeoutContextRef.set( - ReceiveTimeout.ReceiveTimeoutContext( - Some(timeout), - Some(System.currentTimeMillis()), - Some(message) - ) - ) >> action(message) - } else { - Applicative[F].pure(()) + nowMillis.flatMap { currentTime => + val timeoutTime: Long = timestamp + timeout.toMillis + if (timeoutTime <= currentTime) { + receiveTimeoutContextRef.set( + ReceiveTimeout.ReceiveTimeoutContext( + Some(timeout), + Some(currentTime), + Some(message) + ) + ) >> action(message) + } else { + Applicative[F].pure(()) + } } case _ => Applicative[F].pure(()) } diff --git a/shared/src/test/scala/com/suprnation/actor/timeout/ReceiveTimeoutSpec.scala b/shared/src/test/scala/com/suprnation/actor/timeout/ReceiveTimeoutSpec.scala index a237c82..9915ddc 100644 --- a/shared/src/test/scala/com/suprnation/actor/timeout/ReceiveTimeoutSpec.scala +++ b/shared/src/test/scala/com/suprnation/actor/timeout/ReceiveTimeoutSpec.scala @@ -16,6 +16,7 @@ package com.suprnation.actor.timeout +import cats.effect.testkit.TestControl import cats.effect.{IO, Ref} import com.suprnation.actor.Actor.ReplyingReceive import com.suprnation.actor._ @@ -198,4 +199,55 @@ class ReceiveTimeoutSpec extends CatsActorFlatSpec { b1 should be(List(2, 1)) } } + + // Regression test for the Clock-based ReceiveTimeout (see ReceiveTimeout.scala). Under + // cats.effect.testkit.TestControl the receive timeout must fire in *virtual* time, with no real + // waiting: TestControl advances IO.sleep instantly. Before the fix (System.currentTimeMillis), + // virtual time advanced but the wall clock did not, so the timeout never fired here. + it should "fire a receive timeout under TestControl (virtual time, no real waiting)" in { + val program = for { + counter <- Ref.of[IO, Int](0) + buffer <- Ref.of[IO, List[Int]](List.empty) + result <- ActorSystem[IO]("TestControlTimeout", (_: Any) => IO.unit).use { system => + for { + helloActor <- system.replyingActorOf( + constantFlowActor(1 second, counter, buffer), + name = "tc-timeout-actor" + ) + _ <- IO.sleep(2 seconds) // advanced instantly by TestControl, no real wall-clock wait + result1 <- helloActor ? Get + } yield result1 + } + } yield result + + // executeEmbed runs the whole program in simulated time and fails on non-termination. + TestControl.executeEmbed(program).map { case (timeouts, msgs) => + timeouts should be >= 1 + msgs should be(List.empty) + } + } + + // A test that was impossible before this change: a long (1 hour) receive timeout. With a real + // clock this could only be verified by actually waiting an hour; under TestControl the hour of + // IO.sleep is advanced virtually, so it runs in milliseconds and is fully deterministic. + it should "fire a 1-hour receive timeout in virtual time (a real-clock test would wait an hour)" in { + val program = for { + counter <- Ref.of[IO, Int](0) + buffer <- Ref.of[IO, List[Int]](List.empty) + result <- ActorSystem[IO]("LongVirtualTimeout", (_: Any) => IO.unit).use { system => + for { + helloActor <- system.replyingActorOf( + constantFlowActor(1 hour, counter, buffer), + name = "long-timeout-actor" + ) + _ <- IO.sleep(65 minutes) // > 1 hour of virtual time, advanced instantly + result1 <- helloActor ? Get + } yield result1 + } + } yield result + + TestControl.executeEmbed(program).map { case (timeouts, _) => + timeouts should be >= 1 + } + } }