Proposal for ACS ingestion begin

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

@@ -33,8 +33,6 @@ abstract class SpliceAppAutomationService[Store <: AppStore](
     override val store: Store,
     ledgerClient: SpliceLedgerClient,
     retryProvider: RetryProvider,
-    ingestFromParticipantBegin: Boolean,
-    ingestUpdateHistoryFromParticipantBegin: Boolean,
 )(implicit
     ec: ExecutionContext,
     mat: Materializer,
@@ -67,7 +65,6 @@ abstract class SpliceAppAutomationService[Store <: AppStore](
       backoffClock = triggerContext.pollingClock,
       triggerContext.retryProvider,
       triggerContext.loggerFactory,
-      ingestFromParticipantBegin,
     )
   )
 
@@ -80,7 +77,6 @@ abstract class SpliceAppAutomationService[Store <: AppStore](
       backoffClock = triggerContext.pollingClock,
       triggerContext.retryProvider,
       triggerContext.loggerFactory,
-      ingestUpdateHistoryFromParticipantBegin,
     )
   )
 

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

@@ -33,7 +33,6 @@ class UpdateIngestionService(
     backoffClock: Clock,
     override protected val retryProvider: RetryProvider,
     baseLoggerFactory: NamedLoggerFactory,
-    ingestFromParticipantBegin: Boolean,
 )(implicit
     ec: ExecutionContext,
     mat: Materializer,
@@ -59,30 +58,26 @@ class UpdateIngestionService(
           for {
             _ <- ingestAcsAndInFlight(offset)
           } yield offset
+        case IngestionStart.InitializeAtParticipantBegin =>
+          for {
+            participantBegin <- connection.latestPrunedOffset()
+            _ = logger.debug(
+              s"Starting ingestion from participant begin at $participantBegin"
+            )
+            _ <- ingestionSink
+              .ingestAcs(
+                participantBegin,
+                Seq.empty,
+                Seq.empty,
+                Seq.empty,
+              )
+          } yield participantBegin
         case IngestionStart.InitializeAcsAtLatestOffset =>
           for {
-            offset <-
-              if (ingestFromParticipantBegin) {
-                for {
-                  participantBegin <- connection.latestPrunedOffset()
-                  _ = logger.debug(
-                    s"Starting ingestion from participant begin at $participantBegin"
-                  )
-                  _ <- ingestionSink
-                    .ingestAcs(
-                      participantBegin,
-                      Seq.empty,
-                      Seq.empty,
-                      Seq.empty,
-                    )
-                } yield participantBegin
-              } else
-                for {
-                  acsOffset <- connection.ledgerEnd()
-                  _ = logger.debug(s"Starting ingestion from ledger end at $acsOffset")
-                  _ <- ingestAcsAndInFlight(acsOffset)
-                } yield acsOffset
-          } yield offset
+            acsOffset <- connection.ledgerEnd()
+            _ = logger.debug(s"Starting ingestion from ledger end at $acsOffset")
+            _ <- ingestAcsAndInFlight(acsOffset)
+          } yield acsOffset
       }
     } yield new SpliceLedgerSubscription(
       source = connection.updates(subscribeFrom, filter),

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

@@ -758,6 +758,10 @@ object MultiDomainAcsStore {
         */
       final case object InitializeAcsAtLatestOffset extends IngestionStart
 
+      /** Ingestion service should not ingest the ACS, and instead start from the participant's begin
+        */
+      final case object InitializeAtParticipantBegin extends IngestionStart
+
       /** Ingestion service should ingest the ACS at the specified offset,
         * then resume ingesting updates from there
         */

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

@@ -276,9 +276,8 @@ class UpdateHistory(
               IngestionStart.ResumeAtOffset(offset)
             case None =>
               logger.info(s"${description()} initialized")
-              // In case the latest offset is not the beginning of the network,
-              // missing updates will be later backfilled using `ScanHistoryBackfillingTrigger`.
-              IngestionStart.InitializeAcsAtLatestOffset
+              // Missing updates will be later backfilled using `ScanHistoryBackfillingTrigger`.
+              IngestionStart.InitializeAtParticipantBegin
           }
         }
       }

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

@@ -203,8 +203,6 @@ class ScanApp(
         store,
         storage,
         acsSnapshotStore,
-        config.ingestFromParticipantBegin,
-        config.ingestUpdateHistoryFromParticipantBegin,
         serviceUserPrimaryParty,
         svName,
         amuletAppParameters.upgradesConfig,

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

@@ -38,8 +38,6 @@ class ScanAutomationService(
     store: ScanStore,
     storage: Storage,
     snapshotStore: AcsSnapshotStore,
-    ingestFromParticipantBegin: Boolean,
-    ingestUpdateHistoryFromParticipantBegin: Boolean,
     svParty: PartyId,
     svName: String,
     upgradesConfig: UpgradesConfig,
@@ -59,8 +57,6 @@ class ScanAutomationService(
       store,
       ledgerClient,
       retryProvider,
-      ingestFromParticipantBegin,
-      ingestUpdateHistoryFromParticipantBegin,
     ) {
   override def companion
       : org.lfdecentralizedtrust.splice.scan.automation.ScanAutomationService.type =

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

@@ -32,8 +32,6 @@ case class ScanAppBackendConfig(
     sequencerAdminClient: FullClientConfig,
     override val automation: AutomationConfig = AutomationConfig(),
     isFirstSv: Boolean = false,
-    ingestFromParticipantBegin: Boolean = true,
-    ingestUpdateHistoryFromParticipantBegin: Boolean = true,
     miningRoundsCacheTimeToLiveOverride: Option[NonNegativeFiniteDuration] = None,
     acsSnapshotPeriodHours: Int = 3,
     enableForcedAcsSnapshots: Boolean = false,

apps/splitwell/src/main/scala/org/lfdecentralizedtrust/splice/splitwell/automation/SplitwellAutomationService.scala

@@ -62,8 +62,6 @@ class SplitwellAutomationService(
       store,
       ledgerClient,
       retryProvider,
-      ingestFromParticipantBegin = true,
-      ingestUpdateHistoryFromParticipantBegin = true,
     ) {
 
   override def companion

apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/automation/SvDsoAutomationService.scala

@@ -79,8 +79,6 @@ class SvDsoAutomationService(
       dsoStore,
       ledgerClient,
       retryProvider,
-      config.ingestFromParticipantBegin,
-      config.ingestUpdateHistoryFromParticipantBegin,
     ) {
 
   override def companion

apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/automation/SvSvAutomationService.scala

@@ -54,8 +54,6 @@ class SvSvAutomationService(
       svStore,
       ledgerClient,
       retryProvider,
-      config.ingestFromParticipantBegin,
-      config.ingestUpdateHistoryFromParticipantBegin,
     ) {
   override def companion: org.lfdecentralizedtrust.splice.sv.automation.SvSvAutomationService.type =
     SvSvAutomationService

apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/config/SvAppConfig.scala

@@ -293,8 +293,6 @@ case class SvAppBackendConfig(
     onLedgerStatusReportInterval: NonNegativeFiniteDuration =
       NonNegativeFiniteDuration.ofMinutes(2),
     parameters: SpliceParametersConfig = SpliceParametersConfig(batching = BatchingConfig()),
-    ingestFromParticipantBegin: Boolean = true,
-    ingestUpdateHistoryFromParticipantBegin: Boolean = true,
     extraBeneficiaries: Seq[BeneficiaryConfig] = Seq.empty,
     enableOnboardingParticipantPromotionDelay: Boolean = true,
     onboardingPollingInterval: Option[NonNegativeFiniteDuration],

apps/validator/src/main/scala/org/lfdecentralizedtrust/splice/validator/ValidatorApp.scala

@@ -768,8 +768,6 @@ class ValidatorApp(
             loggerFactory,
             domainMigrationInfo,
             participantId,
-            config.ingestFromParticipantBegin,
-            config.ingestUpdateHistoryFromParticipantBegin,
           )
           val walletManager = new UserWalletManager(
             ledgerClient,
@@ -788,8 +786,6 @@ class ValidatorApp(
             loggerFactory,
             domainMigrationInfo,
             participantId,
-            config.ingestFromParticipantBegin,
-            config.ingestUpdateHistoryFromParticipantBegin,
             validatorTopupConfig,
             config.walletSweep,
             config.autoAcceptTransfers,
@@ -831,8 +827,6 @@ class ValidatorApp(
         config.domainMigrationDumpPath,
         config.domainMigrationId,
         retryProvider,
-        config.ingestFromParticipantBegin,
-        config.ingestUpdateHistoryFromParticipantBegin,
         config.svValidator,
         config.sequencerRequestAmplificationPatience,
         config.contactPoint,

apps/validator/src/main/scala/org/lfdecentralizedtrust/splice/validator/automation/ValidatorAutomationService.scala

@@ -63,8 +63,6 @@ class ValidatorAutomationService(
     domainMigrationDumpPath: Option[Path],
     domainMigrationId: Long,
     retryProvider: RetryProvider,
-    ingestFromParticipantBegin: Boolean,
-    ingestUpdateHistoryFromParticipantBegin: Boolean,
     svValidator: Boolean,
     sequencerSubmissionAmplificationPatience: NonNegativeFiniteDuration,
     contactPoint: String,
@@ -83,8 +81,6 @@ class ValidatorAutomationService(
       store,
       ledgerClient,
       retryProvider,
-      ingestFromParticipantBegin,
-      ingestUpdateHistoryFromParticipantBegin,
     ) {
   override def companion
       : org.lfdecentralizedtrust.splice.validator.automation.ValidatorAutomationService.type =

apps/validator/src/main/scala/org/lfdecentralizedtrust/splice/validator/config/ValidatorAppConfig.scala

@@ -171,8 +171,6 @@ case class ValidatorAppBackendConfig(
     // TODO(DACH-NY/canton-network-node#9731): get migration id from sponsor sv / scan instead of configuring here
     domainMigrationId: Long = 0L,
     parameters: SpliceParametersConfig = SpliceParametersConfig(),
-    ingestFromParticipantBegin: Boolean = true,
-    ingestUpdateHistoryFromParticipantBegin: Boolean = true,
     enableWallet: Boolean = true,
     sequencerRequestAmplificationPatience: NonNegativeFiniteDuration =
       ValidatorAppBackendConfig.DEFAULT_SEQUENCER_REQUEST_AMPLIFICATION_PATIENCE,

apps/wallet/src/main/scala/org/lfdecentralizedtrust/splice/wallet/ExternalPartyWalletManager.scala

@@ -40,8 +40,6 @@ class ExternalPartyWalletManager(
     override val loggerFactory: NamedLoggerFactory,
     domainMigrationInfo: DomainMigrationInfo,
     participantId: ParticipantId,
-    ingestFromParticipantBegin: Boolean,
-    ingestUpdateHistoryFromParticipantBegin: Boolean,
 )(implicit
     ec: ExecutionContext,
     mat: Materializer,
@@ -169,8 +167,6 @@ class ExternalPartyWalletManager(
       partyLoggerFactory,
       domainMigrationInfo,
       participantId,
-      ingestFromParticipantBegin,
-      ingestUpdateHistoryFromParticipantBegin,
     )
     (externalPartyRetryProvider, walletService)
   }

apps/wallet/src/main/scala/org/lfdecentralizedtrust/splice/wallet/ExternalPartyWalletService.scala

@@ -36,8 +36,6 @@ class ExternalPartyWalletService(
     override val loggerFactory: NamedLoggerFactory,
     domainMigrationInfo: DomainMigrationInfo,
     participantId: ParticipantId,
-    ingestFromParticipantBegin: Boolean,
-    ingestUpdateHistoryFromParticipantBegin: Boolean,
 )(implicit
     ec: ExecutionContext,
     mat: Materializer,
@@ -67,8 +65,6 @@ class ExternalPartyWalletService(
     domainTimeSync,
     domainUnpausedSync,
     retryProvider,
-    ingestFromParticipantBegin,
-    ingestUpdateHistoryFromParticipantBegin,
     loggerFactory,
   )
 

apps/wallet/src/main/scala/org/lfdecentralizedtrust/splice/wallet/UserWalletManager.scala

@@ -60,8 +60,6 @@ class UserWalletManager(
     override val loggerFactory: NamedLoggerFactory,
     domainMigrationInfo: DomainMigrationInfo,
     participantId: ParticipantId,
-    ingestFromParticipantBegin: Boolean,
-    ingestUpdateHistoryFromParticipantBegin: Boolean,
     validatorTopupConfig: ValidatorTopupConfig,
     walletSweep: Map[String, WalletSweepConfig],
     autoAcceptTransfers: Map[String, AutoAcceptTransfersConfig],
@@ -235,8 +233,6 @@ class UserWalletManager(
       packageVersionSupport,
       domainMigrationInfo,
       participantId,
-      ingestFromParticipantBegin,
-      ingestUpdateHistoryFromParticipantBegin,
       Option.when(endUserParty == store.walletKey.validatorParty)(validatorTopupConfig),
       // TODO(DACH-NY/canton-network-node#12554): make it easier to configure the sweep functionality and guard better against operator errors (typos, etc.)
       walletSweep.get(endUserParty.toProtoPrimitive),

apps/wallet/src/main/scala/org/lfdecentralizedtrust/splice/wallet/UserWalletService.scala

@@ -49,8 +49,6 @@ class UserWalletService(
     packageVersionSupport: PackageVersionSupport,
     domainMigrationInfo: DomainMigrationInfo,
     participantId: ParticipantId,
-    ingestFromParticipantBegin: Boolean,
-    ingestUpdateHistoryFromParticipantBegin: Boolean,
     validatorTopupConfigO: Option[ValidatorTopupConfig],
     walletSweep: Option[WalletSweepConfig],
     autoAcceptTransfers: Option[AutoAcceptTransfersConfig],
@@ -104,8 +102,6 @@ class UserWalletService(
     scanConnection,
     retryProvider,
     packageVersionSupport,
-    ingestFromParticipantBegin,
-    ingestUpdateHistoryFromParticipantBegin,
     loggerFactory,
     validatorTopupConfigO,
     walletSweep,

apps/wallet/src/main/scala/org/lfdecentralizedtrust/splice/wallet/automation/ExternalPartyWalletAutomationService.scala

@@ -30,8 +30,6 @@ class ExternalPartyWalletAutomationService(
     domainTimeSync: DomainTimeSynchronization,
     domainUnpausedSync: DomainUnpausedSynchronization,
     retryProvider: RetryProvider,
-    ingestFromParticipantBegin: Boolean,
-    ingestUpdateHistoryFromParticipantBegin: Boolean,
     override protected val loggerFactory: NamedLoggerFactory,
 )(implicit
     ec: ExecutionContext,
@@ -45,8 +43,6 @@ class ExternalPartyWalletAutomationService(
       store,
       ledgerClient,
       retryProvider,
-      ingestFromParticipantBegin,
-      ingestUpdateHistoryFromParticipantBegin,
     ) {
   override def companion
       : org.lfdecentralizedtrust.splice.wallet.automation.ExternalPartyWalletAutomationService.type =

apps/wallet/src/main/scala/org/lfdecentralizedtrust/splice/wallet/automation/UserWalletAutomationService.scala

@@ -43,8 +43,6 @@ class UserWalletAutomationService(
     scanConnection: BftScanConnection,
     retryProvider: RetryProvider,
     packageVersionSupport: PackageVersionSupport,
-    ingestFromParticipantBegin: Boolean,
-    ingestUpdateHistoryFromParticipantBegin: Boolean,
     override protected val loggerFactory: NamedLoggerFactory,
     validatorTopupConfigO: Option[ValidatorTopupConfig],
     walletSweep: Option[WalletSweepConfig],
@@ -64,8 +62,6 @@ class UserWalletAutomationService(
       store,
       ledgerClient,
       retryProvider,
-      ingestFromParticipantBegin,
-      ingestUpdateHistoryFromParticipantBegin,
     ) {
   override def companion
       : org.lfdecentralizedtrust.splice.wallet.automation.UserWalletAutomationService.type =

cluster/helm/splice-sv-node/templates/sv.yaml

@@ -120,11 +120,6 @@ spec:
             - name: SPLICE_APP_VALIDATOR_AUTH_JWKS_READ_TIMEOUT
               value: {{ .Values.auth.jwks.readTimeout | quote }}
             {{ end }}
-            {{ if .Values.disableIngestUpdateHistoryFromParticipantBegin }}
-            - name: ADDITIONAL_CONFIG_UPDATE_HISTORY_INGESTION
-              value: |
-                canton.sv-apps.sv.ingest-update-history-from-participant-begin = false
-            {{ end }}
             {{- with .Values.persistence }}
             - name: ADDITIONAL_CONFIG_PERSISTENCE
               value: |

cluster/helm/splice-validator/templates/validator.yaml

@@ -257,11 +257,6 @@ spec:
               {{- end }}
             }
         {{- end }}
-        {{ if .Values.disableIngestUpdateHistoryFromParticipantBegin }}
-        - name: ADDITIONAL_CONFIG_UPDATE_HISTORY_INGESTION
-          value: |
-            canton.validator-apps.validator_backend.ingest-update-history-from-participant-begin = false
-        {{ end }}
         - name: FAIL_ON_APP_VERSION_MISMATCH
           value: {{ .Values.failOnAppVersionMismatch | quote }}
         {{- include "splice-util-lib.additional-env-vars" .Values.additionalEnvVars | indent 8}}

docs/src/release_notes.rst

@@ -34,6 +34,11 @@ Upcoming
      - Increase resource requests from 1 CPU and 1Gi to 2 CPUs and 2Gi, to better fit observed resource usage.
      - Remove CPU limits to avoid throttling because of the way K8s handles CPU limits
 
+  - Splice apps
+
+     - Removed the `ingest-from-participant-begin` and `ingest-update-history-from-participant-begin` from splice apps,
+       and the corresponding `disableIngestUpdateHistoryFromParticipantBegin` from helm charts.
+
 0.4.13
 ------