Add feature flag for import update backfilling (#1165)

Signed-off-by: Robert Autenrieth <[email protected]>

Author: rautenrieth-da

apps/app/src/main/scala/org/lfdecentralizedtrust/splice/config/ConfigTransforms.scala

@@ -194,6 +194,7 @@ object ConfigTransforms {
       disableOnboardingParticipantPromotionDelay(),
       setDefaultGrpcDeadlineForBuyExtraTraffic(),
       setDefaultGrpcDeadlineForTreasuryService(),
+      enableImportUpdateBackfilling(),
     )
   }
 
@@ -861,4 +862,10 @@ object ConfigTransforms {
     rows.toMap
   }
 
+  def enableImportUpdateBackfilling(): ConfigTransform =
+    updateAllScanAppConfigs((_, config) =>
+      config
+        .copy(updateHistoryBackfillImportUpdatesEnabled = true)
+    )
+
 }

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

@@ -77,6 +77,7 @@ class UpdateHistory(
     val backfillingRequired: BackfillingRequirement,
     override protected val loggerFactory: NamedLoggerFactory,
     enableissue12777Workaround: Boolean,
+    enableImportUpdateBackfill: Boolean,
     val oMetrics: Option[HistoryMetrics] = None,
 )(implicit
     ec: ExecutionContext,
@@ -245,7 +246,13 @@ class UpdateHistory(
 
           _ <- cleanUpDataAfterDomainMigration(newHistoryId)
 
-          _ <- deleteInvalidAcsSnapshots(newHistoryId)
+          _ <-
+            if (enableImportUpdateBackfill) {
+              deleteInvalidAcsSnapshots(newHistoryId)
+            } else {
+              logger.info(s"Not deleting invalid ACS snapshots for history $newHistoryId")
+              Future.unit
+            }
         } yield {
           state.updateAndGet(
             _.copy(
@@ -649,6 +656,7 @@ class UpdateHistory(
   private[this] def deleteInvalidAcsSnapshots(
       historyId: Long
   )(implicit tc: TraceContext): Future[Unit] = {
+    assert(enableImportUpdateBackfill)
     def migrationsWithCorruptSnapshots(): Future[Set[Long]] = {
       for {
         migrationsWithImportUpdates <- storage
@@ -1834,7 +1842,8 @@ class UpdateHistory(
 
   def getBackfillingState()(implicit
       tc: TraceContext
-  ): Future[BackfillingState] = getBackfillingStateForHistory(historyId)
+  ): Future[BackfillingState] =
+    getBackfillingStateForHistory(historyId)
 
   private[this] def getBackfillingStateForHistory(historyId: Long)(implicit
       tc: TraceContext
@@ -1856,6 +1865,9 @@ class UpdateHistory(
             case Some((updatesComplete, importUpdatesComplete)) =>
               if (updatesComplete && importUpdatesComplete) {
                 BackfillingState.Complete
+              } else if (updatesComplete && !enableImportUpdateBackfill) {
+                // If import update backfilling is disabled, behave as if it was not implemented
+                BackfillingState.Complete
               } else {
                 BackfillingState.InProgress(
                   updatesComplete = updatesComplete,

apps/common/src/main/scala/org/lfdecentralizedtrust/splice/store/db/DbAppStore.scala

@@ -24,6 +24,7 @@ abstract class DbTxLogAppStore[TXE](
     domainMigrationInfo: DomainMigrationInfo,
     participantId: ParticipantId,
     enableissue12777Workaround: Boolean,
+    enableImportUpdateBackfill: Boolean,
     backfillingRequired: BackfillingRequirement,
     oHistoryMetrics: Option[HistoryMetrics] = None,
 )(implicit
@@ -37,6 +38,7 @@ abstract class DbTxLogAppStore[TXE](
       domainMigrationInfo = domainMigrationInfo,
       participantId = participantId,
       enableissue12777Workaround = enableissue12777Workaround,
+      enableImportUpdateBackfill = enableImportUpdateBackfill,
       backfillingRequired,
       oHistoryMetrics = oHistoryMetrics,
     )
@@ -66,6 +68,7 @@ abstract class DbAppStore(
     domainMigrationInfo: DomainMigrationInfo,
     participantId: ParticipantId,
     enableissue12777Workaround: Boolean,
+    enableImportUpdateBackfill: Boolean,
     backfillingRequired: BackfillingRequirement,
     oHistoryMetrics: Option[HistoryMetrics] = None,
 )(implicit
@@ -112,6 +115,7 @@ abstract class DbAppStore(
       backfillingRequired,
       loggerFactory,
       enableissue12777Workaround,
+      enableImportUpdateBackfill,
       oHistoryMetrics,
     )
 

apps/common/src/test/scala/org/lfdecentralizedtrust/splice/store/TxLogBackfillingStoreTest.scala

@@ -554,6 +554,7 @@ class TxLogBackfillingStoreTest
       backfillingRequired,
       loggerFactory,
       enableissue12777Workaround = true,
+      enableImportUpdateBackfill = true,
     )
   }
 

apps/common/src/test/scala/org/lfdecentralizedtrust/splice/store/UpdateHistoryTestBase.scala

@@ -252,6 +252,7 @@ abstract class UpdateHistoryTestBase
       backfillingRequired,
       loggerFactory,
       enableissue12777Workaround = true,
+      enableImportUpdateBackfill = true,
     )
   }
 

apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/ScanApp.scala

@@ -172,6 +172,7 @@ class ScanApp(
         migrationInfo,
         participantId,
         config.cache.svNodeStateTtl,
+        config.updateHistoryBackfillImportUpdatesEnabled,
         nodeMetrics.dbScanStore,
       )
       acsSnapshotStore = AcsSnapshotStore(

apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/automation/ScanAutomationService.scala

@@ -71,6 +71,7 @@ class ScanAutomationService(
         svName,
         ledgerClient,
         config.updateHistoryBackfillBatchSize,
+        config.updateHistoryBackfillImportUpdatesEnabled,
         svParty,
         upgradesConfig,
         triggerContext,

apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/automation/ScanHistoryBackfillingTrigger.scala

@@ -49,6 +49,7 @@ class ScanHistoryBackfillingTrigger(
     svName: String,
     ledgerClient: SpliceLedgerClient,
     batchSize: Int,
+    importUpdateBackfillingEnabled: Boolean,
     svParty: PartyId,
     upgradesConfig: UpgradesConfig,
     override protected val context: TriggerContext,
@@ -122,6 +123,11 @@ class ScanHistoryBackfillingTrigger(
     case ScanHistoryBackfillingTrigger.InitializeBackfillingTask(_) =>
       initializeBackfilling()
     case ScanHistoryBackfillingTrigger.ImportUpdatesBackfillTask() =>
+      if (!importUpdateBackfillingEnabled) {
+        throw new RuntimeException(
+          "Import updates backfilling is disabled, this place should not be reached"
+        )
+      }
       performImportUpdatesBackfilling()
     case ScanHistoryBackfillingTrigger.BackfillTask() =>
       performBackfilling()

apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/config/ScanAppConfig.scala

@@ -43,10 +43,12 @@ case class ScanAppBackendConfig(
     spliceInstanceNames: SpliceInstanceNamesConfig,
     updateHistoryBackfillEnabled: Boolean = true,
     updateHistoryBackfillBatchSize: Int = 100,
+    updateHistoryBackfillImportUpdatesEnabled: Boolean = false,
     txLogBackfillEnabled: Boolean = true,
     txLogBackfillBatchSize: Int = 100,
     bftSequencers: Seq[BftSequencerConfig] = Seq.empty,
     cache: ScanCacheConfig = ScanCacheConfig(),
+    // TODO(#1164): Enable by default
 ) extends SpliceBackendConfig
     with BaseScanAppConfig // TODO(DACH-NY/canton-network-node#736): fork or generalize this trait.
     {

apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/ScanStore.scala

@@ -319,6 +319,7 @@ object ScanStore {
       domainMigrationInfo: DomainMigrationInfo,
       participantId: ParticipantId,
       svNodeStateCacheTtl: NonNegativeFiniteDuration,
+      enableImportUpdateBackfill: Boolean,
       metrics: DbScanStoreMetrics,
   )(implicit
       ec: ExecutionContext,
@@ -337,6 +338,7 @@ object ScanStore {
           domainMigrationInfo,
           participantId,
           svNodeStateCacheTtl,
+          enableImportUpdateBackfill,
           metrics,
         )
       case storageType => throw new RuntimeException(s"Unsupported storage type $storageType")