Cache SV node state based queries (#1150)

moritzkiefer-da · cocreature · web-flow · commit 97286a4f9ce7 · 2025-06-18T09:02:34.000Z
* Cache SV node state based queries [ci] fixes #1141 Signed-off-by: Moritz Kiefer <moritz.kiefer@purelyfunctional.org> * fix cache metrics [ci] Signed-off-by: Moritz Kiefer <moritz.kiefer@purelyfunctional.org> * review [ci] Signed-off-by: Moritz Kiefer <moritz.kiefer@purelyfunctional.org> * fix [ci] Signed-off-by: Moritz Kiefer <moritz.kiefer@purelyfunctional.org> --------- Signed-off-by: Moritz Kiefer <moritz.kiefer@purelyfunctional.org> Co-authored-by: Moritz Kiefer <moritz.kiefer@purelyfunctional.org>
diff --git a/apps/app/src/main/scala/org/lfdecentralizedtrust/splice/config/SpliceConfig.scala b/apps/app/src/main/scala/org/lfdecentralizedtrust/splice/config/SpliceConfig.scala
@@ -15,6 +15,7 @@ import org.lfdecentralizedtrust.splice.scan.config.{
   BftSequencerConfig,
   ScanAppBackendConfig,
   ScanAppClientConfig,
+  ScanCacheConfig,
   ScanSynchronizerConfig,
 }
 import org.lfdecentralizedtrust.splice.splitwell.config.{
@@ -427,6 +428,8 @@ object SpliceConfig {
       ConfigReader.forProduct3("migration-id", "sequencer-admin-client", "p2p-url")(
         BftSequencerConfig(_, _, _)
       )
+    implicit val scanCacheConfigReader: ConfigReader[ScanCacheConfig] =
+      deriveReader[ScanCacheConfig]
     implicit val scanConfigReader: ConfigReader[ScanAppBackendConfig] =
       deriveReader[ScanAppBackendConfig]
 
@@ -794,6 +797,8 @@ object SpliceConfig {
       )
     implicit val scanConfigWriter: ConfigWriter[ScanAppBackendConfig] =
       deriveWriter[ScanAppBackendConfig]
+    implicit val scanCacheConfigWriter: ConfigWriter[ScanCacheConfig] =
+      deriveWriter[ScanCacheConfig]
 
     implicit val svClientConfigWriter: ConfigWriter[SvAppClientConfig] =
       deriveWriter[SvAppClientConfig]
diff --git a/apps/app/src/test/resources/include/scans/_scan.conf b/apps/app/src/test/resources/include/scans/_scan.conf
@@ -12,4 +12,6 @@
   participant-client {
     ledger-api.auth-config.type = "none"
   }
+  # Reduce in tests as SVs change relatively frequently here.
+  cache.sv-node-state-ttl = 1s
 }
diff --git a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/ScanApp.scala b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/ScanApp.scala
@@ -171,6 +171,7 @@ class ScanApp(
         },
         migrationInfo,
         participantId,
+        config.cache.svNodeStateTtl,
         nodeMetrics.dbScanStore,
       )
       acsSnapshotStore = AcsSnapshotStore(
diff --git a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/admin/api/client/BftScanConnection.scala b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/admin/api/client/BftScanConnection.scala
@@ -1033,7 +1033,7 @@ object BftScanConnection {
     ): Future[Seq[DsoScan]] = {
       for {
         decentralizedSynchronizerId <- store.getDecentralizedSynchronizerId()
-        scans <- store.listDsoScans()
+        scans <- store.listCachedDsoScans()
         domainScans <- scans
           .find(_._1 == decentralizedSynchronizerId.toProtoPrimitive)
           .map(e => Future.successful(e._2.filter(_.svName != ownSvName)))
diff --git a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/admin/http/HttpScanHandler.scala b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/admin/http/HttpScanHandler.scala
@@ -610,14 +610,13 @@ class HttpScanHandler(
       .map(ScanResource.ListValidatorLicensesResponse.OK)
   }
 
-  // TODO: (#7809) Add caching for sequencers per domain
   override def listDsoSequencers(
       respond: v0.ScanResource.ListDsoSequencersResponse.type
   )()(extracted: TraceContext): Future[v0.ScanResource.ListDsoSequencersResponse] = {
     implicit val tc = extracted
     withSpan(s"$workflowId.listDsoSequencers") { _ => _ =>
       store
-        .listFromSvNodeStates { nodeState =>
+        .listFromCachedSvNodeStates { nodeState =>
           for {
             (synchronizerId, domainConfig) <- nodeState.state.synchronizerNodes.asScala.toVector
             sequencers = for {
@@ -670,7 +669,7 @@ class HttpScanHandler(
     implicit val tc: TraceContext = extracted
     withSpan(s"$workflowId.listDsoScans") { _ => _ =>
       store
-        .listDsoScans()
+        .listCachedDsoScans()
         .map(list =>
           definitions.ListDsoScansResponse(list.map { case (synchronizerId, scans) =>
             definitions.DomainScans(
diff --git a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/config/ScanAppConfig.scala b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/config/ScanAppConfig.scala
@@ -46,6 +46,7 @@ case class ScanAppBackendConfig(
     txLogBackfillEnabled: Boolean = true,
     txLogBackfillBatchSize: Int = 100,
     bftSequencers: Seq[BftSequencerConfig] = Seq.empty,
+    cache: ScanCacheConfig = ScanCacheConfig(),
 ) extends SpliceBackendConfig
     with BaseScanAppConfig // TODO(DACH-NY/canton-network-node#736): fork or generalize this trait.
     {
@@ -54,6 +55,10 @@ case class ScanAppBackendConfig(
   override def clientAdminApi: ClientConfig = adminApi.clientConfig
 }
 
+final case class ScanCacheConfig(
+    svNodeStateTtl: NonNegativeFiniteDuration = NonNegativeFiniteDuration.ofSeconds(30)
+)
+
 case class ScanAppClientConfig(
     adminApi: NetworkAppClientConfig,
 
diff --git a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/ScanStore.scala b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/ScanStore.scala
@@ -36,6 +36,7 @@ import org.lfdecentralizedtrust.splice.util.{
   ContractWithState,
   TemplateJsonDecoder,
 }
+import com.digitalasset.canton.config.NonNegativeFiniteDuration
 import com.digitalasset.canton.data.CantonTimestamp
 import com.digitalasset.canton.lifecycle.CloseContext
 import com.digitalasset.canton.logging.NamedLoggerFactory
@@ -97,26 +98,29 @@ trait ScanStore
       )
     )
 
+  protected def listCachedSvNodeStates()(implicit
+      tc: TraceContext
+  ): Future[Seq[splice.dso.svstate.SvNodeState]]
+
   /** Returns all items extracted by `f` from the DsoRules ensuring that they're sorted by synchronizerId,
     * so that the order is deterministic.
     */
-  def listFromSvNodeStates[T](
+  def listFromCachedSvNodeStates[T](
       f: splice.dso.svstate.SvNodeState => Vector[(String, T)]
   )(implicit tc: TraceContext): Future[Vector[(String, Vector[T])]] = {
     for {
-      dsoRules <- getDsoRulesWithState()
-      nodeStates <- Future.traverse(dsoRules.payload.svs.asScala.keys) { svPartyId =>
-        getSvNodeState(PartyId.tryFromProtoPrimitive(svPartyId))
-      }
+      nodeStates <- listCachedSvNodeStates()
     } yield {
-      val items = nodeStates.toVector.flatMap(nodeState => f(nodeState.contract.payload))
+      val items = nodeStates.toVector.flatMap(nodeState => f(nodeState))
       val itemsByDomain = items.groupBy(_._1).view.mapValues(_.map(_._2))
       itemsByDomain.toVector.sortBy(_._1)
     }
   }
 
-  def listDsoScans()(implicit tc: TraceContext): Future[Vector[(String, Vector[ScanInfo])]] = {
-    listFromSvNodeStates { nodeState =>
+  def listCachedDsoScans()(implicit
+      tc: TraceContext
+  ): Future[Vector[(String, Vector[ScanInfo])]] = {
+    listFromCachedSvNodeStates { nodeState =>
       for {
         (synchronizerId, domainConfig) <- nodeState.state.synchronizerNodes.asScala.toVector
         scan <- domainConfig.scan.toScala
@@ -314,6 +318,7 @@ object ScanStore {
       createScanAggregatesReader: DbScanStore => ScanAggregatesReader,
       domainMigrationInfo: DomainMigrationInfo,
       participantId: ParticipantId,
+      svNodeStateCacheTtl: NonNegativeFiniteDuration,
       metrics: DbScanStoreMetrics,
   )(implicit
       ec: ExecutionContext,
@@ -331,6 +336,7 @@ object ScanStore {
           createScanAggregatesReader,
           domainMigrationInfo,
           participantId,
+          svNodeStateCacheTtl,
           metrics,
         )
       case storageType => throw new RuntimeException(s"Unsupported storage type $storageType")
diff --git a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/db/DbScanStore.scala b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/db/DbScanStore.scala
@@ -5,7 +5,7 @@ package org.lfdecentralizedtrust.splice.scan.store.db
 
 import com.daml.ledger.javaapi.data.codegen.ContractId
 import com.digitalasset.canton.caching.ScaffeineCache
-import com.digitalasset.canton.config.NonNegativeDuration
+import com.digitalasset.canton.config.{NonNegativeDuration, NonNegativeFiniteDuration}
 import com.digitalasset.canton.data.CantonTimestamp
 import com.digitalasset.canton.lifecycle.{
   AsyncCloseable,
@@ -83,6 +83,7 @@ import org.lfdecentralizedtrust.splice.store.db.TxLogQueries.TxLogStoreId
 
 import java.time.Instant
 import scala.concurrent.{ExecutionContext, Future}
+import scala.jdk.CollectionConverters.*
 
 object DbScanStore {
   type CacheKey = java.lang.Long // caffeine metrics function demands AnyRefs
@@ -97,6 +98,7 @@ class DbScanStore(
     createScanAggregatesReader: DbScanStore => ScanAggregatesReader,
     domainMigrationInfo: DomainMigrationInfo,
     participantId: ParticipantId,
+    svNodeStateCacheTtl: NonNegativeFiniteDuration,
     storeMetrics: DbScanStoreMetrics,
 )(implicit
     override protected val ec: ExecutionContext,
@@ -587,14 +589,40 @@ class DbScanStore(
       Scaffeine()
         .maximumSize(1000),
       key => getUncachedTotalAmuletBalance(key),
-      metrics = Some(storeMetrics.cache),
+      metrics = Some(storeMetrics.totalAmuletBalanceCache),
     )(logger, "amuletBalanceCache")
   }
   // TODO(#800) remove when amulet expiry works again
   def getTotalAmuletBalance(asOfEndOfRound: Long): Future[BigDecimal] = {
     totalAmuletBalanceCache.get(asOfEndOfRound)
   }
 
+  private val svNodeStateCache: ScaffeineCache.TunnelledAsyncLoadingCache[
+    Future,
+    Unit,
+    Seq[SvNodeState],
+  ] = {
+    implicit val tc = TraceContext.empty
+    ScaffeineCache.buildAsync[Future, Unit, Seq[SvNodeState]](
+      Scaffeine()
+        .expireAfterWrite(svNodeStateCacheTtl.asFiniteApproximation),
+      _ => this.listSvNodeStates(),
+      metrics = Some(storeMetrics.svNodeStateCache),
+    )(logger, "svNodeStateCache")
+  }
+
+  private def listSvNodeStates()(implicit tc: TraceContext): Future[Seq[SvNodeState]] =
+    for {
+      dsoRules <- getDsoRulesWithState()
+      nodeStates <- Future.traverse(dsoRules.payload.svs.asScala.keys) { svPartyId =>
+        getSvNodeState(PartyId.tryFromProtoPrimitive(svPartyId))
+      }
+    } yield nodeStates.map(_.contract.payload).toVector
+
+  protected override def listCachedSvNodeStates()(implicit
+      tc: TraceContext
+  ): Future[Seq[SvNodeState]] = svNodeStateCache.get(())
+
   protected override def getUncachedTotalAmuletBalance(asOfEndOfRound: Long)(implicit
       tc: TraceContext
   ): Future[BigDecimal] =
diff --git a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/db/DbScanStoreMetrics.scala b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/db/DbScanStoreMetrics.scala
@@ -36,7 +36,9 @@ class DbScanStoreMetrics(metricsFactory: LabeledMetricsFactory) extends AutoClos
       -1L,
     )(MetricsContext.Empty)
 
-  val cache = new CacheMetrics(prefix :+ "cache", metricsFactory)
+  val totalAmuletBalanceCache =
+    new CacheMetrics(prefix :+ "total_amulet_balance_cache", metricsFactory)
+  val svNodeStateCache = new CacheMetrics(prefix :+ "sv_node_state_cache", metricsFactory)
 
   val history = new HistoryMetrics(metricsFactory)(MetricsContext.Empty)
 
diff --git a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/db/ScanAggregatesReader.scala b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/db/ScanAggregatesReader.scala
@@ -131,7 +131,7 @@ object ScanAggregatesReader {
         decentralizedSynchronizerId <- store
           .getDecentralizedSynchronizerId()
           .map(_.unwrap.toProtoPrimitive)
-        scans <- store.listDsoScans()
+        scans <- store.listCachedDsoScans()
         scanUrls = scans
           .filter(_._1 == decentralizedSynchronizerId)
           .flatMap { case (_, scans) =>
diff --git a/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/store/db/ScanAggregatorTest.scala b/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/store/db/ScanAggregatorTest.scala
@@ -18,6 +18,7 @@ import org.lfdecentralizedtrust.splice.util.ResourceTemplateDecoder
 import org.lfdecentralizedtrust.splice.util.TemplateJsonDecoder
 import com.digitalasset.canton.HasExecutionContext
 import com.digitalasset.canton.concurrent.FutureSupervisor
+import com.digitalasset.canton.config.NonNegativeFiniteDuration
 import com.digitalasset.canton.data.CantonTimestamp
 import com.daml.metrics.api.noop.NoOpMetricsFactory
 import com.digitalasset.canton.resource.DbStorage
@@ -695,6 +696,7 @@ class ScanAggregatorTest
         None,
       ),
       participantId = mkParticipantId("ScanAggregatorTest"),
+      svNodeStateCacheTtl = NonNegativeFiniteDuration.ofSeconds(30),
       new DbScanStoreMetrics(new NoOpMetricsFactory()),
     )(parallelExecutionContext, implicitly, implicitly)
     for {
diff --git a/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/store/db/ScanStoreTest.scala b/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/store/db/ScanStoreTest.scala
@@ -70,6 +70,7 @@ import org.lfdecentralizedtrust.splice.util.{
   TemplateJsonDecoder,
 }
 import com.digitalasset.canton.concurrent.FutureSupervisor
+import com.digitalasset.canton.config.NonNegativeFiniteDuration
 import com.digitalasset.canton.crypto.Fingerprint
 import com.digitalasset.canton.data.CantonTimestamp
 import com.daml.metrics.api.noop.NoOpMetricsFactory
@@ -2329,6 +2330,7 @@ class DbScanStoreTest
         None,
       ),
       participantId = mkParticipantId("ScanStoreTest"),
+      svNodeStateCacheTtl = NonNegativeFiniteDuration.ofSeconds(30),
       new DbScanStoreMetrics(new NoOpMetricsFactory()),
     )(parallelExecutionContext, implicitly, implicitly)
 
diff --git a/canton/community/base/src/main/scala/com/digitalasset/canton/caching/ScaffeineCache.scala b/canton/community/base/src/main/scala/com/digitalasset/canton/caching/ScaffeineCache.scala
@@ -29,7 +29,9 @@ object ScaffeineCache {
       context: String,
   ): TunnelledAsyncCache[K, V] = {
     val contextWithLoggerName = s"${tracedLogger.underlying.getName}:$context"
-    val asyncCache = cache.buildAsync[K, V]()
+    val cacheWithMetrics =
+      metrics.fold(cache)(m => cache.recordStats(() => new StatsCounterMetrics(m)))
+    val asyncCache = cacheWithMetrics.buildAsync[K, V]()
     metrics.foreach(CaffeineCache.installMetrics(_, asyncCache.underlying.synchronous()))
     new TunnelledAsyncCache[K, V](
       asyncCache,
@@ -51,7 +53,9 @@ object ScaffeineCache {
     // when the tunnelled effect gets logged by `com.github.benmanes.caffeine.cache.LocalAsyncCache.handleCompletion`,
     // which bypasses our logger factories.
     val contextWithLoggerName = s"${tracedLogger.underlying.getName}:$context"
-    val asyncLoadingCache = cache.buildAsyncFuture[K, V](
+    val cacheWithMetrics =
+      metrics.fold(cache)(m => cache.recordStats(() => new StatsCounterMetrics(m)))
+    val asyncLoadingCache = cacheWithMetrics.buildAsyncFuture[K, V](
       loader = key => tunnel.enter(loader(key), contextWithLoggerName),
       allLoader = allLoader.map(loader => keys => tunnel.enter(loader(keys), contextWithLoggerName)),
     )

Original file line number	Diff line number	Diff line change
`@@ -12,4 +12,6 @@`
`12`	`12`	`participant-client {`
`13`	`13`	`ledger-api.auth-config.type = "none"`
`14`	`14`	`}`
	`15`	`+ # Reduce in tests as SVs change relatively frequently here.`
	`16`	`+ cache.sv-node-state-ttl = 1s`
`15`	`17`	`}`
Original file line number	Diff line number	Diff line change
`@@ -171,6 +171,7 @@ class ScanApp(`
`171`	`171`	`},`
`172`	`172`	`migrationInfo,`
`173`	`173`	`participantId,`
	`174`	`+ config.cache.svNodeStateTtl,`
`174`	`175`	`nodeMetrics.dbScanStore,`
`175`	`176`	`)`
`176`	`177`	`acsSnapshotStore = AcsSnapshotStore(`