[ci] add test

OriolMunoz-da · OriolMunoz-da · commit 403f5de46e66 · 2026-01-19T17:02:32.000Z
Signed-off-by: Oriol Muñoz &lt;oriol.munoz@digitalasset.com&gt;
diff --git a/apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/tests/TokenStandardTransferIntegrationTest.scala b/apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/tests/TokenStandardTransferIntegrationTest.scala
@@ -8,20 +8,20 @@ import org.lfdecentralizedtrust.splice.codegen.java.splice.api.token.metadatav1
 import org.lfdecentralizedtrust.splice.codegen.java.splice.api.token.transferinstructionv1.TransferInstruction
 import org.lfdecentralizedtrust.splice.config.ConfigTransforms.{
   ConfigurableApp,
+  updateAllScanAppConfigs_,
   updateAllSvAppFoundDsoConfigs_,
   updateAutomationConfig,
 }
-import org.lfdecentralizedtrust.splice.http.v0.definitions.TransferInstructionResultOutput.members
 import org.lfdecentralizedtrust.splice.http.v0.definitions.TransactionHistoryResponseItem.TransactionType as HttpTransactionType
+import org.lfdecentralizedtrust.splice.http.v0.definitions.TransferInstructionResultOutput.members
 import org.lfdecentralizedtrust.splice.http.v0.definitions.{
   AbortTransferInstruction,
   ReceiverAmount,
   Transfer,
 }
 import org.lfdecentralizedtrust.splice.integration.EnvironmentDefinition
 import org.lfdecentralizedtrust.splice.integration.tests.SpliceTests.IntegrationTestWithSharedEnvironment
-import org.lfdecentralizedtrust.splice.integration.tests.WalletTxLogTestUtil
-import org.lfdecentralizedtrust.splice.util.{DisclosedContracts, WalletTestUtil}
+import org.lfdecentralizedtrust.splice.util.{DisclosedContracts, TriggerTestUtil, WalletTestUtil}
 import org.lfdecentralizedtrust.splice.wallet.automation.CollectRewardsAndMergeAmuletsTrigger
 import org.lfdecentralizedtrust.splice.wallet.store.{
   BalanceChangeTxLogEntry,
@@ -39,6 +39,7 @@ class TokenStandardTransferIntegrationTest
     extends IntegrationTestWithSharedEnvironment
     with WalletTestUtil
     with WalletTxLogTestUtil
+    with TriggerTestUtil
     with HasActorSystem
     with HasExecutionContext {
 
@@ -56,10 +57,56 @@ class TokenStandardTransferIntegrationTest
           _.copy(zeroTransferFees = true)
         )(config)
       )
+      .addConfigTransforms((_, config) =>
+        updateAllScanAppConfigs_(scanConfig =>
+          scanConfig.copy(parameters =
+            scanConfig.parameters.copy(contractFetchLedgerFallbackEnabled = true)
+          )
+        )(config)
+      )
   }
 
   "Token Standard Transfers should" should {
 
+    "TransferInstruction context can be fetched from Scan even if it's not yet ingested into the store" in {
+      implicit env =>
+        pauseScanIngestionWithin(sv1ScanBackend) {
+          onboardWalletUser(aliceWalletClient, aliceValidatorBackend)
+          val bobUserParty = onboardWalletUser(bobWalletClient, bobValidatorBackend)
+          aliceWalletClient.tap(100)
+
+          val response = actAndCheck(
+            "Alice creates transfer offer",
+            aliceWalletClient.createTokenStandardTransfer(
+              bobUserParty,
+              10,
+              s"Transfer",
+              CantonTimestamp.now().plusSeconds(3600L),
+              UUID.randomUUID().toString,
+            ),
+          )(
+            "Alice and Bob see it",
+            _ => {
+              Seq(aliceWalletClient, bobWalletClient).foreach(
+                _.listTokenStandardTransfers() should have size 1
+              )
+            },
+          )._1
+
+          val cid = response.output match {
+            case members.TransferInstructionPending(value) =>
+              new TransferInstruction.ContractId(value.transferInstructionCid)
+            case _ => fail("The transfers were expected to be pending.")
+          }
+
+          clue("SV-1's Scan sees it (stiil, even though ingestion is paused)") {
+            eventuallySucceeds() {
+              sv1ScanBackend.getTransferInstructionAcceptContext(cid)
+            }
+          }
+        }
+    }
+
     "support create, list, accept, reject and withdraw" in { implicit env =>
       val aliceUserParty = onboardWalletUser(aliceWalletClient, aliceValidatorBackend)
       val bobUserParty = onboardWalletUser(bobWalletClient, bobValidatorBackend)
diff --git a/apps/app/src/test/scala/org/lfdecentralizedtrust/splice/util/TriggerTestUtil.scala b/apps/app/src/test/scala/org/lfdecentralizedtrust/splice/util/TriggerTestUtil.scala
@@ -2,7 +2,8 @@ package org.lfdecentralizedtrust.splice.util
 
 import com.digitalasset.canton.{BaseTest, ScalaFuturesWithPatience}
 import com.typesafe.scalalogging.LazyLogging
-import org.lfdecentralizedtrust.splice.automation.Trigger
+import org.lfdecentralizedtrust.splice.automation.{Trigger, UpdateIngestionService}
+import org.lfdecentralizedtrust.splice.console.ScanAppBackendReference
 import org.lfdecentralizedtrust.splice.integration.EnvironmentDefinition.sv1Backend
 import org.lfdecentralizedtrust.splice.integration.tests.SpliceTests.SpliceTestConsoleEnvironment
 import org.lfdecentralizedtrust.splice.sv.automation.delegatebased.AdvanceOpenMiningRoundTrigger
@@ -34,6 +35,17 @@ trait TriggerTestUtil { self: BaseTest =>
       advanceOpenMiningRoundTrigger.runOnce().futureValue should be(true)
     }
   }
+
+  def pauseScanIngestionWithin[T](scan: ScanAppBackendReference)(codeBlock: => T): T = {
+    try {
+      logger.info(s"Pausing ingestion for ${scan.name}")
+      scan.automation.services[UpdateIngestionService].foreach(_.pause().futureValue)
+      codeBlock
+    } finally {
+      logger.info(s"Resuming ingestion for ${scan.name}")
+      scan.automation.services[UpdateIngestionService].foreach(_.resume())
+    }
+  }
 }
 
 object TriggerTestUtil extends ScalaFuturesWithPatience with LazyLogging {
diff --git a/apps/common/src/main/scala/org/lfdecentralizedtrust/splice/automation/AutomationService.scala b/apps/common/src/main/scala/org/lfdecentralizedtrust/splice/automation/AutomationService.scala
@@ -83,11 +83,13 @@ abstract class AutomationService(
   }
 
   /** Returns all triggers of the given class */
-  final def triggers[T <: Trigger](implicit tag: ClassTag[T]): Seq[T] = {
+  final def triggers[T <: Trigger](implicit tag: ClassTag[T]): Seq[T] = services[T](tag)
+
+  final def services[T](implicit tag: ClassTag[T]): Seq[T] = {
     backgroundServices
       .get()
-      .collect { case trigger: T =>
-        trigger
+      .collect { case service: T =>
+        service
       }
   }
 
diff --git a/apps/common/src/main/scala/org/lfdecentralizedtrust/splice/automation/UpdateIngestionService.scala b/apps/common/src/main/scala/org/lfdecentralizedtrust/splice/automation/UpdateIngestionService.scala
@@ -19,7 +19,7 @@ import com.digitalasset.canton.tracing.TraceContext
 import io.opentelemetry.api.trace.Tracer
 import org.lfdecentralizedtrust.splice.store.MultiDomainAcsStore.IngestionSink.IngestionStart
 
-import scala.concurrent.{ExecutionContext, Future}
+import scala.concurrent.{ExecutionContext, Future, Promise, blocking}
 
 /** Ingestion for ACS and transfer stores.
   * We ingest them independently but we ensure that the acs store
@@ -97,13 +97,18 @@ class UpdateIngestionService(
   private def process(
       msgs: Seq[GetTreeUpdatesResponse]
   )(implicit traceContext: TraceContext): Future[Unit] = {
-    NonEmptyList.fromFoldable(msgs) match {
-      case Some(batch) =>
-        logger.debug(s"Processing batch of ${batch.size} elements")
-        ingestionSink.ingestUpdateBatch(batch.map(_.updateOrCheckpoint))
-      case None =>
-        logger.error("Received empty batch of updates to ingest. This is never supposed to happen.")
-        Future.unit
+    // if paused, this step will backpressure the source
+    waitForResumePromise.future.flatMap { _ =>
+      NonEmptyList.fromFoldable(msgs) match {
+        case Some(batch) =>
+          logger.debug(s"Processing batch of ${batch.size} elements")
+          ingestionSink.ingestUpdateBatch(batch.map(_.updateOrCheckpoint))
+        case None =>
+          logger.error(
+            "Received empty batch of updates to ingest. This is never supposed to happen."
+          )
+          Future.unit
+      }
     }
   }
 
@@ -124,4 +129,36 @@ class UpdateIngestionService(
 
   // Kick-off the ingestion
   startIngestion()
+
+  @SuppressWarnings(Array("org.wartremover.warts.Var"))
+  @volatile
+  private var waitForResumePromise = Promise.successful(())
+
+  /** Note that any in-flight events being processed when `pause` is called will still be processed.
+    * For test purposes.
+    */
+  def pause(): Future[Unit] = blocking {
+    withNewTrace(this.getClass.getSimpleName) { implicit traceContext => _ =>
+      logger.info("Pausing UpdateIngestionService.")
+      blocking {
+        synchronized {
+          if (waitForResumePromise.isCompleted) {
+            waitForResumePromise = Promise()
+          }
+          Future.successful(())
+        }
+      }
+    }
+  }
+
+  def resume(): Unit = blocking {
+    withNewTrace(this.getClass.getSimpleName) { implicit traceContext => _ =>
+      logger.info("Resuming UpdateIngestionService.")
+      blocking {
+        synchronized {
+          val _ = waitForResumePromise.trySuccess(())
+        }
+      }
+    }
+  }
 }
diff --git a/apps/common/src/main/scala/org/lfdecentralizedtrust/splice/store/ContractFetcher.scala b/apps/common/src/main/scala/org/lfdecentralizedtrust/splice/store/ContractFetcher.scala
@@ -52,7 +52,7 @@ object ContractFetcher {
       OptionT(store.multiDomainAcsStore.lookupContractById(companion)(id))
         .map(_.contract)
         .orElse(
-          OptionT(fallbackLedgerClient.getContract(id, Seq(store.multiDomainAcsStore.storePartyId)))
+          OptionT(fallbackLedgerClient.getContract(id, Seq(store.multiDomainAcsStore.storeParty)))
             .subflatMap { createdEvent =>
               companionClass
                 .fromCreatedEvent(companion)(CreatedEvent.fromProto(toJavaProto(createdEvent)))
diff --git a/apps/common/src/main/scala/org/lfdecentralizedtrust/splice/store/MultiDomainAcsStore.scala b/apps/common/src/main/scala/org/lfdecentralizedtrust/splice/store/MultiDomainAcsStore.scala
@@ -52,12 +52,11 @@ import scala.jdk.CollectionConverters.*
 
 trait MultiDomainAcsStore extends HasIngestionSink with AutoCloseable with NamedLogging {
   protected def storeName: String
-  protected def storeParty: String
-  def storePartyId = PartyId.tryFromProtoPrimitive(storeParty)
+  def storeParty: PartyId
 
   protected implicit lazy val mc: MetricsContext = MetricsContext(
     "store_name" -> storeName,
-    "store_party" -> storeParty,
+    "store_party" -> storeParty.toString, // using .toString for historical reasons
   )
   protected def metricsFactory: LabeledMetricsFactory
 
diff --git a/apps/common/src/main/scala/org/lfdecentralizedtrust/splice/store/db/DbMultiDomainAcsStore.scala b/apps/common/src/main/scala/org/lfdecentralizedtrust/splice/store/db/DbMultiDomainAcsStore.scala
@@ -115,7 +115,7 @@ final class DbMultiDomainAcsStore[TXE](
   import profile.api.jdbcActionExtensionMethods
 
   override lazy val storeName = acsStoreDescriptor.name
-  override lazy val storeParty = acsStoreDescriptor.party.toString
+  override lazy val storeParty = acsStoreDescriptor.party
 
   override protected def metricsFactory: LabeledMetricsFactory = retryProvider.metricsFactory
   override lazy val metrics = new StoreMetrics(metricsFactory)(mc)

Original file line number	Diff line number	Diff line change
`@@ -83,11 +83,13 @@ abstract class AutomationService(`
`83`	`83`	`}`
`84`	`84`
`85`	`85`	`/** Returns all triggers of the given class */`
`86`		`- final def triggers[T <: Trigger](implicit tag: ClassTag[T]): Seq[T] = {`
	`86`	`+ final def triggers[T <: Trigger](implicit tag: ClassTag[T]): Seq[T] = services[T](tag)`
	`87`	`+`
	`88`	`+ final def services[T](implicit tag: ClassTag[T]): Seq[T] = {`
`87`	`89`	`backgroundServices`
`88`	`90`	`.get()`
`89`		`- .collect { case trigger: T =>`
`90`		`- trigger`
	`91`	`+ .collect { case service: T =>`
	`92`	`+ service`
`91`	`93`	`}`
`92`	`94`	`}`
`93`	`95`