Disable UpdateHistory for SV and Splitwell apps

apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/plugins/UpdateHistorySanityCheckPlugin.scala

@@ -247,7 +247,7 @@ class UpdateHistorySanityCheckPlugin(
             interval = Span(100, Millis),
           )
         eventually {
-          scan.automation.store.updateHistory
+          scan.automation.updateHistory
             .getBackfillingState()
             .futureValue should be(BackfillingState.Complete)
         }(

apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/tests/DecentralizedSynchronizerMigrationIntegrationTest.scala

@@ -1093,14 +1093,14 @@ class DecentralizedSynchronizerMigrationIntegrationTest
 
             withClueAndLog("Backfilled history includes ACS import") {
               eventually() {
-                sv1ScanLocalBackend.appState.store.updateHistory.sourceHistory
+                sv1ScanLocalBackend.appState.automation.updateHistory.sourceHistory
                   .migrationInfo(1L)
                   .futureValue
                   .exists(_.complete) should be(true)
               }
 
               val backfilledUpdates =
-                sv1ScanLocalBackend.appState.store.updateHistory
+                sv1ScanLocalBackend.appState.automation.updateHistory
                   .getAllUpdates(None, PageLimit.tryCreate(1000))
                   .futureValue
               backfilledUpdates.collect {

apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/tests/ScanHistoryBackfillingIntegrationTest.scala

@@ -281,8 +281,8 @@ class ScanHistoryBackfillingIntegrationTest
     )(
       "History marked as free of corrupt snapshots",
       _ => {
-        sv1ScanBackend.appState.store.updateHistory.corruptAcsSnapshotsDeleted shouldBe true
-        sv2ScanBackend.appState.store.updateHistory.corruptAcsSnapshotsDeleted shouldBe true
+        sv1ScanBackend.appState.automation.updateHistory.corruptAcsSnapshotsDeleted shouldBe true
+        sv2ScanBackend.appState.automation.updateHistory.corruptAcsSnapshotsDeleted shouldBe true
       },
     )
 
@@ -309,15 +309,15 @@ class ScanHistoryBackfillingIntegrationTest
     )(
       "Backfilling is complete only on the founding SV",
       _ => {
-        sv1ScanBackend.appState.store.updateHistory
+        sv1ScanBackend.appState.automation.updateHistory
           .getBackfillingState()
           .futureValue should be(BackfillingState.Complete)
         // Update history is complete at this point, but the status endpoint only reports
         // as complete if the txlog is also backfilled
         sv1ScanBackend.getBackfillingStatus().complete shouldBe false
         readUpdateHistoryFromScan(sv1ScanBackend) should not be empty
 
-        sv2ScanBackend.appState.store.updateHistory
+        sv2ScanBackend.appState.automation.updateHistory
           .getBackfillingState()
           .futureValue should be(BackfillingState.InProgress(false, false))
         sv2ScanBackend.getBackfillingStatus().complete shouldBe false
@@ -356,12 +356,12 @@ class ScanHistoryBackfillingIntegrationTest
     )(
       "All backfilling is complete",
       _ => {
-        sv1ScanBackend.appState.store.updateHistory
+        sv1ScanBackend.appState.automation.updateHistory
           .getBackfillingState()
           .futureValue should be(BackfillingState.Complete)
         // Update history is complete, TxLog is not
         sv1ScanBackend.getBackfillingStatus().complete shouldBe false
-        sv2ScanBackend.appState.store.updateHistory
+        sv2ScanBackend.appState.automation.updateHistory
           .getBackfillingState()
           .futureValue should be(BackfillingState.Complete)
         // Update history is complete, TxLog is not
@@ -446,7 +446,7 @@ class ScanHistoryBackfillingIntegrationTest
     clue("Compare scan history with participant update stream") {
       compareHistory(
         sv1Backend.participantClient,
-        sv1ScanBackend.appState.store.updateHistory,
+        sv1ScanBackend.appState.automation.updateHistory,
         ledgerBeginSv1,
       )
     }
@@ -556,7 +556,7 @@ class ScanHistoryBackfillingIntegrationTest
 
   private def allUpdatesFromScanBackend(scanBackend: ScanAppBackendReference) = {
     // Need to use the store directly, as the HTTP endpoint refuses to return data unless it's completely backfilled
-    scanBackend.appState.store.updateHistory
+    scanBackend.appState.automation.updateHistory
       .getAllUpdates(None, PageLimit.tryCreate(1000))
       .futureValue
   }

apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/tests/ScanTimeBasedIntegrationTest.scala

@@ -434,7 +434,7 @@ class ScanTimeBasedIntegrationTest
       "Wait for backfilling to complete, as the ACS snapshot trigger is paused until then"
     ) {
       eventually() {
-        sv1ScanBackend.automation.store.updateHistory
+        sv1ScanBackend.automation.updateHistory
           .getBackfillingState()
           .futureValue should be(BackfillingState.Complete)
         advanceTime(sv1ScanBackend.config.automation.pollingInterval.asJava)

apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/tests/ScanTotalSupplyBigQueryIntegrationTest.scala

@@ -314,7 +314,7 @@ class ScanTotalSupplyBigQueryIntegrationTest
       case db: DbStorage => db
       case s => fail(s"non-DB storage configured, unsupported for BigQuery: ${s.getClass}")
     }
-    val sourceHistoryId = sv1ScanBackend.appState.store.updateHistory.historyId
+    val sourceHistoryId = sv1ScanBackend.appState.automation.updateHistory.historyId
 
     copyTableToBigQuery(
       "update_history_creates",

apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/tests/UpdateHistoryIntegrationTest.scala

@@ -156,7 +156,7 @@ class UpdateHistoryIntegrationTest
       eventually() {
         compareHistory(
           sv1Backend.participantClient,
-          sv1ScanBackend.appState.store.updateHistory,
+          sv1ScanBackend.appState.automation.updateHistory,
           ledgerBeginSv1,
         )
       }
@@ -181,30 +181,16 @@ class UpdateHistoryIntegrationTest
             .lookupUserWallet(aliceWalletClient.config.ledgerApiUser)
             .futureValue
             .getOrElse(throw new RuntimeException("Alice wallet should exist"))
-            .store
+            .automation
             .updateHistory,
           ledgerBeginAlice,
           true,
         )
       }
-      eventually() {
-        compareHistory(
-          sv1Backend.participantClient,
-          sv1Backend.appState.svStore.updateHistory,
-          ledgerBeginSv1,
-        )
-      }
-      eventually() {
-        compareHistory(
-          sv1Backend.participantClient,
-          sv1Backend.appState.dsoStore.updateHistory,
-          ledgerBeginSv1,
-        )
-      }
       eventually() {
         compareHistory(
           aliceValidatorBackend.participantClient,
-          aliceValidatorBackend.appState.store.updateHistory,
+          aliceValidatorBackend.appState.automation.updateHistory,
           ledgerBeginAlice,
         )
       }

apps/app/src/test/scala/org/lfdecentralizedtrust/splice/util/UpdateHistoryTestUtil.scala

@@ -137,7 +137,7 @@ trait UpdateHistoryTestUtil extends TestCommon {
       scanBackend: ScanAppBackendReference,
       scanClient: ScanAppClientReference,
   ): Assertion = {
-    val historyFromStore = scanBackend.appState.store.updateHistory
+    val historyFromStore = scanBackend.appState.automation.updateHistory
       .getAllUpdates(
         None,
         PageLimit.tryCreate(1000),

apps/common/src/main/scala/org/lfdecentralizedtrust/splice/automation/AutomationService.scala

@@ -23,7 +23,7 @@ import scala.reflect.ClassTag
 
 /** Shared base class for running ingestion and task-handler automation in applications. */
 abstract class AutomationService(
-    private val automationConfig: AutomationConfig,
+    protected val automationConfig: AutomationConfig,
     clock: Clock,
     domainTimeSync: DomainTimeSynchronization,
     domainUnpausedSync: DomainUnpausedSynchronization,

apps/common/src/main/scala/org/lfdecentralizedtrust/splice/automation/SpliceAppAutomationService.scala

@@ -37,7 +37,6 @@ abstract class SpliceAppAutomationService[Store <: AppStore](
     ledgerClient: SpliceLedgerClient,
     retryProvider: RetryProvider,
     ingestFromParticipantBegin: Boolean,
-    ingestUpdateHistoryFromParticipantBegin: Boolean,
     parametersConfig: SpliceParametersConfig,
 )(implicit
     ec: ExecutionContext,
@@ -113,19 +112,6 @@ abstract class SpliceAppAutomationService[Store <: AppStore](
     )
   )
 
-  registerService(
-    new UpdateIngestionService(
-      store.updateHistory.getClass.getSimpleName,
-      store.updateHistory.ingestionSink,
-      connection(SpliceLedgerConnectionPriority.High),
-      automationConfig,
-      backoffClock = triggerContext.pollingClock,
-      triggerContext.retryProvider,
-      triggerContext.loggerFactory,
-      ingestUpdateHistoryFromParticipantBegin,
-    )
-  )
-
   registerTrigger(
     new DomainIngestionService(
       store.domains.ingestionSink,

apps/common/src/main/scala/org/lfdecentralizedtrust/splice/automation/TxLogBackfillingTrigger.scala

@@ -10,6 +10,7 @@ import org.lfdecentralizedtrust.splice.store.{
   HistoryMetrics,
   TxLogAppStore,
   TxLogBackfilling,
+  UpdateHistory,
 }
 import com.digitalasset.canton.logging.pretty.{Pretty, PrettyPrinting}
 import com.digitalasset.canton.topology.PartyId
@@ -23,6 +24,7 @@ import scala.concurrent.{ExecutionContext, Future}
 
 class TxLogBackfillingTrigger[TXE](
     store: TxLogAppStore[TXE],
+    updateHistory: UpdateHistory,
     batchSize: Int,
     override protected val context: TriggerContext,
 )(implicit
@@ -31,13 +33,13 @@ class TxLogBackfillingTrigger[TXE](
     mat: Materializer,
 ) extends PollingParallelTaskExecutionTrigger[TxLogBackfillingTrigger.Task] {
 
-  private def party: PartyId = store.updateHistory.updateStreamParty
+  private def party: PartyId = updateHistory.updateStreamParty
 
   override protected def extraMetricLabels = Seq(
     "party" -> party.toProtoPrimitive
   )
 
-  private val currentMigrationId = store.updateHistory.domainMigrationInfo.currentMigrationId
+  private val currentMigrationId = updateHistory.domainMigrationInfo.currentMigrationId
 
   private val historyMetrics = new HistoryMetrics(context.metricsFactory)(
     MetricsContext.Empty
@@ -48,23 +50,23 @@ class TxLogBackfillingTrigger[TXE](
   )
   private val backfilling = new TxLogBackfilling(
     store.multiDomainAcsStore,
-    store.updateHistory,
+    updateHistory,
     batchSize,
     context.loggerFactory,
   )
 
   override def retrieveTasks()(implicit
       tc: TraceContext
   ): Future[Seq[TxLogBackfillingTrigger.Task]] = {
-    if (!store.updateHistory.isReady) {
+    if (!updateHistory.isReady) {
       logger.debug("UpdateHistory is not yet ready")
       Future.successful(Seq.empty)
     } else if (!store.multiDomainAcsStore.destinationHistory.isReady) {
       logger.debug("MultiDomainAcsStore is not yet ready")
       Future.successful(Seq.empty)
     } else {
       for {
-        sourceState <- store.updateHistory.getBackfillingState()
+        sourceState <- updateHistory.getBackfillingState()
         destinationState <- store.multiDomainAcsStore.getTxLogBackfillingState()
       } yield {
         sourceState match {

apps/common/src/main/scala/org/lfdecentralizedtrust/splice/environment/NodeBootstrapBase.scala

@@ -14,7 +14,7 @@ import com.digitalasset.canton.crypto.Crypto
 import com.digitalasset.canton.environment.{CantonNode, CantonNodeBootstrap, CantonNodeParameters}
 import com.digitalasset.canton.lifecycle.{HasCloseContext, LifeCycle}
 import com.digitalasset.canton.logging.NamedLoggerFactory
-import com.digitalasset.canton.resource.StorageFactory
+import com.digitalasset.canton.resource.{DbStorage, StorageFactory}
 import com.digitalasset.canton.telemetry.ConfiguredOpenTelemetry
 import com.digitalasset.canton.time.Clock
 import com.digitalasset.canton.tracing.{NoTracing, TracerProvider}
@@ -102,7 +102,10 @@ abstract class NodeBootstrapBase[
         nodeMetrics.storageMetrics,
         parameterConfig.processingTimeouts,
         loggerFactory,
-      )
+      ) match {
+      case storage: DbStorage => storage
+      case storageType => throw new RuntimeException(s"Unsupported storage type $storageType")
+    }
   protected val httpAdminService: HttpAdminService =
     HttpAdminService(
       nodeConfig.nodeTypeName,

apps/common/src/main/scala/org/lfdecentralizedtrust/splice/store/AppStore.scala

@@ -12,17 +12,18 @@ import scala.concurrent.ExecutionContext
   */
 trait AppStore extends NamedLogging with AutoCloseable with StoreErrors {
 
+  val storeName: String
+
   implicit protected def ec: ExecutionContext
 
   /** Defines which create events are to be ingested into the store. */
-  protected def acsContractFilter
+  def acsContractFilter
       : MultiDomainAcsStore.ContractFilter[_ <: AcsRowData, _ <: AcsInterfaceViewRowData]
 
   def domains: SynchronizerStore
 
   def multiDomainAcsStore: MultiDomainAcsStore
 
-  def updateHistory: UpdateHistory
 }
 
 trait TxLogAppStore[TXE] extends AppStore {

apps/common/src/main/scala/org/lfdecentralizedtrust/splice/store/HistoryMetrics.scala

@@ -209,3 +209,9 @@ class HistoryMetrics(metricsFactory: LabeledMetricsFactory)(implicit
     ImportUpdatesBackfilling.completed.close()
   }
 }
+
+object HistoryMetrics {
+  def apply(metricsFactory: LabeledMetricsFactory, currentMigrationId: Long) = new HistoryMetrics(
+    metricsFactory
+  )(MetricsContext("current_migration_id" -> currentMigrationId.toString))
+}

apps/common/src/main/scala/org/lfdecentralizedtrust/splice/store/UpdateHistory.scala

@@ -96,14 +96,15 @@ class UpdateHistory(
     override protected val loggerFactory: NamedLoggerFactory,
     enableissue12777Workaround: Boolean,
     enableImportUpdateBackfill: Boolean,
-    val oMetrics: Option[HistoryMetrics] = None,
+    metrics: HistoryMetrics,
 )(implicit
     ec: ExecutionContext,
     closeContext: CloseContext,
 ) extends HasIngestionSink
     with AcsJdbcTypes
     with AcsQueries
-    with NamedLogging {
+    with NamedLogging
+    with AutoCloseable {
 
   override lazy val profile: JdbcProfile = storage.api.jdbcProfile
 
@@ -133,6 +134,8 @@ class UpdateHistory(
 
   def isReady: Boolean = state.get().historyId.isDefined
 
+  override def close(): Unit = metrics.close()
+
   lazy val ingestionSink: MultiDomainAcsStore.IngestionSink =
     new MultiDomainAcsStore.IngestionSink {
       override def ingestionFilter: IngestionFilter = IngestionFilter(
@@ -442,7 +445,7 @@ class UpdateHistory(
     val safeParticipantOffset = lengthLimited(LegacyOffset.Api.fromLong(reassignment.offset))
     val safeUnassignId = lengthLimited(event.unassignId)
     val safeContractId = lengthLimited(event.contractId.contractId)
-    oMetrics.foreach(_.UpdateHistory.unassignments.mark())
+    metrics.UpdateHistory.unassignments.mark()
     sqlu"""
       insert into update_history_unassignments(
         history_id,update_id,record_time,
@@ -488,7 +491,7 @@ class UpdateHistory(
     val safeCreatedAt = CantonTimestamp.assertFromInstant(event.createdEvent.createdAt)
     val safeSignatories = event.createdEvent.getSignatories.asScala.toSeq.map(lengthLimited)
     val safeObservers = event.createdEvent.getObservers.asScala.toSeq.map(lengthLimited)
-    oMetrics.foreach(_.UpdateHistory.assignments.mark())
+    metrics.UpdateHistory.assignments.mark()
     sqlu"""
       insert into update_history_assignments(
         history_id,update_id,record_time,
@@ -518,7 +521,7 @@ class UpdateHistory(
       tree: TransactionTree,
       migrationId: Long,
   ): DBIOAction[?, NoStream, Effect.Read & Effect.Write] = {
-    oMetrics.foreach(_.UpdateHistory.transactionsTrees.mark())
+    metrics.UpdateHistory.transactionsTrees.mark()
     insertTransactionUpdateRow(tree, migrationId).flatMap(updateRowId => {
       // Note: the order of elements in the eventsById map doesn't matter, and is not preserved here.
       // The order of elements in the rootEventIds and childEventIds lists DOES matter, and needs to be preserved.