ReceiveTimeout: use cats-effect Clock instead of System.currentTimeMillis (TestControl support)

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
     )
   )

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(())
     }

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