make EvalCacheMulti easier to test

Author: ornicar

src/main/scala/evalCache/EvalCacheApi.scala

@@ -1,6 +1,8 @@
 package lila.ws
 package evalCache
 
+import org.apache.pekko.actor.typed.Scheduler
+import java.time.LocalDateTime
 import cats.syntax.all.*
 import chess.ErrorStr
 import chess.format.Fen
@@ -10,19 +12,14 @@ import com.typesafe.scalalogging.Logger
 import play.api.libs.json.JsString
 import reactivemongo.api.bson.BSONDocument
 
-import java.time.LocalDateTime
-
 import lila.ws.ipc.ClientIn
 import lila.ws.ipc.ClientOut.{ EvalGet, EvalGetMulti, EvalPut }
 
-final class EvalCacheApi(mongo: Mongo, settings: util.SettingStore)(using
-    Executor,
-    org.apache.pekko.actor.typed.Scheduler
-):
+final class EvalCacheApi(mongo: Mongo)(using Executor, Scheduler):
 
   private val truster = wire[EvalCacheTruster]
   private val upgrade = wire[EvalCacheUpgrade]
-  private val multi   = EvalCacheMulti()
+  private val multi   = EvalCacheMulti.withMonitoring()
 
   import EvalCacheEntry.*
   import EvalCacheBsonHandlers.given

src/main/scala/evalCache/EvalCacheMulti.scala

@@ -1,30 +1,28 @@
 package lila.ws
 package evalCache
 
+import java.util.concurrent.ConcurrentHashMap
+import org.apache.pekko.actor.typed.Scheduler
 import chess.format.Fen
 import chess.variant.Variant
-
-import java.util.concurrent.ConcurrentHashMap
+import scalalib.zeros.given
 
 import lila.ws.ipc.ClientIn.EvalHitMulti
 import lila.ws.ipc.ClientOut.EvalGetMulti
 import lila.ws.util.ExpireCallbackMemo
+import lila.ws.util.ExpireMemo
 
 /* Compared to EvalCacheUpgrade, accepts multiple positions per member,
  * only sends cp/mate
  */
-final private class EvalCacheMulti(using
-    ec: Executor,
-    scheduler: org.apache.pekko.actor.typed.Scheduler
-):
+final private class EvalCacheMulti private (makeExpirableSris: (Sri => Unit) => ExpireMemo[Sri]):
   import EvalCacheMulti.*
   import EvalCacheUpgrade.{ EvalState, SriString }
 
-  private val members       = ConcurrentHashMap[SriString, WatchingMember](4096)
-  private val evals         = ConcurrentHashMap[Id, EvalState](1024)
-  private val expirableSris = ExpireCallbackMemo[Sri](scheduler, 1 minute, expire)
+  private val members                        = ConcurrentHashMap[SriString, WatchingMember](4096)
+  private val evals                          = ConcurrentHashMap[Id, EvalState](1024)
+  private val expirableSris: ExpireMemo[Sri] = makeExpirableSris(expire)
 
-  private val upgradeMon = Monitor.evalCache.multi.upgrade
 
   def register(sri: Sri, e: EvalGetMulti): Unit =
     members
@@ -41,22 +39,24 @@ final private class EvalCacheMulti(using
     expirableSris.put(sri)
 
   def onEval(input: EvalCacheEntry.Input): Unit =
-    Option(
+    val sris = onEvalSrisToUpgrade(input)
+    if sris.nonEmpty then
+      val hit = EvalHitMulti:
+        EvalCacheJsonHandlers.writeMultiHit(input.fen, input.eval)
+      sris.foreach: sri =>
+        Bus.publish(_.sri(sri), hit)
+      upgradeMon.count.increment(sris.size)
+
+  private def onEvalSrisToUpgrade(input: EvalCacheEntry.Input): Set[Sri] =
+    val newEval = Option(
       evals.computeIfPresent(
         input.id,
         (_, ev) =>
           if ev.depth >= input.eval.depth then ev
           else ev.copy(depth = input.eval.depth)
       )
     ).filter(_.depth == input.eval.depth)
-      .foreach: eval =>
-        val sris = eval.sris.filter(_ != input.sri)
-        if sris.nonEmpty then
-          val hit = EvalHitMulti:
-            EvalCacheJsonHandlers.writeMultiHit(input.fen, input.eval)
-          sris.foreach: sri =>
-            Bus.publish(_.sri(sri), hit)
-          upgradeMon.count.increment(sris.size)
+    newEval.so(_.sris.filter(_ != input.sri))
 
   private def expire(sri: Sri): Unit =
     Option(members.remove(sri.value)).foreach:
@@ -71,14 +71,19 @@ final private class EvalCacheMulti(using
         else eval.copy(sris = newSris)
     )
 
-  scheduler.scheduleWithFixedDelay(1 minute, 1 minute): () =>
-    upgradeMon.members.update(members.size)
-    upgradeMon.evals.update(evals.size)
-    upgradeMon.expirable.update(expirableSris.count)
-
 private object EvalCacheMulti:
 
   import EvalCacheUpgrade.*
 
   case class WatchingMember(sri: Sri, variant: Variant, fens: List[Fen.Full]):
     def setups: List[Id] = fens.flatMap(Id.from(variant, _))
+
+  private val upgradeMon = Monitor.evalCache.multi.upgrade
+
+  def withMonitoring()(using ec: Executor, scheduler: Scheduler): EvalCacheMulti =
+    val instance = EvalCacheMulti(expire => ExpireCallbackMemo[Sri](1 minute, expire))
+    scheduler.scheduleWithFixedDelay(1 minute, 1 minute): () =>
+      upgradeMon.members.update(instance.members.size)
+      upgradeMon.evals.update(instance.evals.size)
+      upgradeMon.expirable.update(instance.expirableSris.count)
+    instance

src/main/scala/evalCache/EvalCacheUpgrade.scala

@@ -1,12 +1,12 @@
 package lila.ws
 package evalCache
 
+import java.util.concurrent.ConcurrentHashMap
+import org.apache.pekko.actor.typed.Scheduler
 import chess.format.UciPath
 import play.api.libs.json.JsString
 import scalalib.DebouncerFunction
 
-import java.util.concurrent.ConcurrentHashMap
-
 import lila.ws.ipc.ClientIn.EvalHit
 import lila.ws.ipc.ClientOut.EvalGet
 import lila.ws.util.ExpireCallbackMemo
@@ -15,15 +15,12 @@ import lila.ws.util.ExpireCallbackMemo
  * by remembering the last evalGet of each socket member,
  * and listening to new evals stored.
  */
-final private class EvalCacheUpgrade(using
-    ec: Executor,
-    scheduler: org.apache.pekko.actor.typed.Scheduler
-):
+final private class EvalCacheUpgrade(using ec: Executor, scheduler: Scheduler):
   import EvalCacheUpgrade.*
 
   private val members       = ConcurrentHashMap[SriString, WatchingMember](4096)
   private val evals         = ConcurrentHashMap[SetupId, EvalState](1024)
-  private val expirableSris = ExpireCallbackMemo[Sri](scheduler, 3 minutes, expire)
+  private val expirableSris = ExpireCallbackMemo[Sri](3 minutes, expire)
 
   private val debouncer = DebouncerFunction[SetupId](scheduler.scheduleOnce(5.seconds, _), 64)
 

src/main/scala/util/ExpireCallbackMemo.scala

@@ -7,13 +7,17 @@ import org.apache.pekko.actor.typed.Scheduler
 import java.util.concurrent.ConcurrentHashMap
 import scala.jdk.CollectionConverters.*
 
+trait ExpireMemo[K]:
+  def put(key: K): Unit
+  def count: Int
+
 // calls a function when a key expires
 final class ExpireCallbackMemo[K](
-    scheduler: Scheduler,
     ttl: FiniteDuration,
     callback: K => Unit,
     initialCapacity: Int = 4096
-)(using Executor):
+)(using ec: Executor, scheduler: Scheduler)
+    extends ExpireMemo[K]:
 
   private val timeouts = ConcurrentHashMap[K, Cancellable](initialCapacity)