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

@@ -279,8 +279,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
       },
     )
 
@@ -307,15 +307,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
@@ -354,12 +354,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
@@ -444,7 +444,7 @@ class ScanHistoryBackfillingIntegrationTest
     clue("Compare scan history with participant update stream") {
       compareHistory(
         sv1Backend.participantClient,
-        sv1ScanBackend.appState.store.updateHistory,
+        sv1ScanBackend.appState.automation.updateHistory,
         ledgerBeginSv1,
       )
     }
@@ -554,7 +554,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

@@ -154,7 +154,7 @@ class UpdateHistoryIntegrationTest
       eventually() {
         compareHistory(
           sv1Backend.participantClient,
-          sv1ScanBackend.appState.store.updateHistory,
+          sv1ScanBackend.appState.automation.updateHistory,
           ledgerBeginSv1,
         )
       }
@@ -179,30 +179,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/resources/db/migration/canton-network/postgres/stable/V047__truncate_sv_app_history.sql

@@ -0,0 +1,26 @@
+-- if the only 2 store descriptors belong to the SV app, that means we can truncate the update_history tables
+
+DO $$
+DECLARE
+    descriptors TEXT[];
+BEGIN
+
+-- array equality (ordered) ensures that exactly these two, no more, no less, are there <=> it's the SV app
+select array_agg(store_name order by store_name) into descriptors
+from update_history_descriptors;
+
+IF descriptors  = '{"DbSvDsoStore", "DbSvSvStore"}' THEN
+    RAISE NOTICE 'Truncating update history tables as only SV app descriptors are present. Descriptors: %', descriptors::text;
+    EXECUTE 'TRUNCATE TABLE update_history_assignments CASCADE';
+    EXECUTE 'TRUNCATE TABLE update_history_unassignments CASCADE';
+    EXECUTE 'TRUNCATE TABLE update_history_backfilling CASCADE';
+    EXECUTE 'TRUNCATE TABLE update_history_creates CASCADE';
+    EXECUTE 'TRUNCATE TABLE update_history_exercises CASCADE';
+    EXECUTE 'TRUNCATE TABLE update_history_transactions CASCADE';
+    EXECUTE 'TRUNCATE TABLE update_history_last_ingested_offsets CASCADE';
+    EXECUTE 'TRUNCATE TABLE update_history_descriptors CASCADE';
+ELSE
+    RAISE NOTICE 'This is not the SV app, NOT truncating update history tables. Descriptors: %', descriptors::text;
+END IF;
+
+END $$;

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

@@ -15,6 +15,7 @@ import org.lfdecentralizedtrust.splice.store.{
   AppStoreWithIngestion,
   DomainTimeSynchronization,
   DomainUnpausedSynchronization,
+  UpdateHistory,
 }
 import com.digitalasset.canton.time.{Clock, WallClock}
 import com.digitalasset.canton.tracing.TraceContext
@@ -37,7 +38,6 @@ abstract class SpliceAppAutomationService[Store <: AppStore](
     ledgerClient: SpliceLedgerClient,
     retryProvider: RetryProvider,
     ingestFromParticipantBegin: Boolean,
-    ingestUpdateHistoryFromParticipantBegin: Boolean,
     parametersConfig: SpliceParametersConfig,
 )(implicit
     ec: ExecutionContext,
@@ -97,6 +97,24 @@ abstract class SpliceAppAutomationService[Store <: AppStore](
       case SpliceLedgerConnectionPriority.AmuletExpiry => amuletExpiryConnection
     }
 
+  final protected def registerUpdateHistoryIngestion(
+      updateHistory: UpdateHistory,
+      ingestUpdateHistoryFromParticipantBegin: Boolean,
+  ): Unit = {
+    registerService(
+      new UpdateIngestionService(
+        updateHistory.getClass.getSimpleName,
+        updateHistory.ingestionSink,
+        connection(SpliceLedgerConnectionPriority.High),
+        automationConfig,
+        backoffClock = triggerContext.pollingClock,
+        triggerContext.retryProvider,
+        triggerContext.loggerFactory,
+        ingestUpdateHistoryFromParticipantBegin,
+      )
+    )
+  }
+
   private def completionOffsetCallback(offset: Long): Future[Unit] =
     store.multiDomainAcsStore.signalWhenIngestedOrShutdown(offset)(TraceContext.empty)
 
@@ -113,19 +131,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/SqlIndexInitializationTrigger.scala

@@ -8,7 +8,7 @@ import com.digitalasset.canton.data.CantonTimestamp
 import com.digitalasset.canton.discard.Implicits.DiscardOps
 import com.digitalasset.canton.lifecycle.{FutureUnlessShutdown, *}
 import com.digitalasset.canton.logging.pretty.{Pretty, PrettyPrinting}
-import com.digitalasset.canton.resource.{DbStorage, Storage}
+import com.digitalasset.canton.resource.DbStorage
 import com.digitalasset.canton.tracing.TraceContext
 import io.opentelemetry.api.trace.Tracer
 import org.apache.pekko.stream.Materializer
@@ -150,23 +150,19 @@ class SqlIndexInitializationTrigger(
 object SqlIndexInitializationTrigger {
 
   def apply(
-      storage: Storage,
+      storage: DbStorage,
       triggerContext: TriggerContext,
       indexActions: List[IndexAction] = defaultIndexActions,
   )(implicit
       ec: ExecutionContextExecutor,
       tracer: Tracer,
       mat: Materializer,
-  ): SqlIndexInitializationTrigger = storage match {
-    case dbStorage: DbStorage =>
-      new SqlIndexInitializationTrigger(
-        dbStorage,
-        triggerContext,
-        indexActions,
-      )
-    case storageType =>
-      // Same behavior as in `ScanStore.apply` and similar - we only really support DbStorage in our apps.
-      throw new RuntimeException(s"Unsupported storage type $storageType")
+  ): SqlIndexInitializationTrigger = {
+    new SqlIndexInitializationTrigger(
+      storage,
+      triggerContext,
+      indexActions,
+    )
   }
 
   sealed trait IndexStatus

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 {