From bca0d2feed6457eb4df31574f7946af092ad15ad Mon Sep 17 00:00:00 2001 From: sadiq1971 Date: Thu, 27 Aug 2026 18:08:48 +0600 Subject: [PATCH 1/3] fix: review wording on the sync operator app [ci] Drop the speculative "yet" from the domain time comment, and call the init step what the variable already calls it. Signed-off-by: sadiq1971 --- .../splice/syncoperator/SyncOperatorApp.scala | 2 +- .../syncoperator/automation/SyncOperatorAutomationService.scala | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/SyncOperatorApp.scala b/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/SyncOperatorApp.scala index 2478c861ad..58f39f47fa 100644 --- a/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/SyncOperatorApp.scala +++ b/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/SyncOperatorApp.scala @@ -132,7 +132,7 @@ class SyncOperatorApp( config.automation.ingestion, config.parameters.defaultLimit, ) - globalSynchronizerId <- appInitStep("Get the decentralized synchronizer id") { + globalSynchronizerId <- appInitStep("Get the global synchronizer id") { scanConnection.getAmuletRulesDomain()(traceContext) } readOnlyLedgerConnection = ledgerClient diff --git a/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/automation/SyncOperatorAutomationService.scala b/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/automation/SyncOperatorAutomationService.scala index e606bac1f3..3821e1e148 100644 --- a/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/automation/SyncOperatorAutomationService.scala +++ b/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/automation/SyncOperatorAutomationService.scala @@ -42,7 +42,7 @@ class SyncOperatorAutomationService( ) extends SpliceAppAutomationService( automationConfig, clock, - // Nothing registered here depends on domain time yet. + // Nothing registered here depends on domain time. DomainTimeSynchronization.Noop, store, ledgerClient, From 6021ca72c6ca3b24748933b996b395c37c05f1eb Mon Sep 17 00:00:00 2001 From: sadiq1971 Date: Thu, 27 Aug 2026 19:58:31 +0600 Subject: [PATCH 2/3] fix: pin the sync operator store's migration id to 0 [ci] MIGRATION_ID is frozen network wide and logical synchronizer upgrades carry a serial id instead, so the store's partition can never move and resolving it bought nothing. Drops resolveDomainMigrationId and its scan fallback. It also lines the store's stamp up with the payload.migrationId == 0L check in the contract filter, so there is no longer a second migration id a few lines away meaning something different. Signed-off-by: sadiq1971 --- .../splice/syncoperator/SyncOperatorApp.scala | 30 ++----------------- 1 file changed, 3 insertions(+), 27 deletions(-) diff --git a/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/SyncOperatorApp.scala b/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/SyncOperatorApp.scala index 58f39f47fa..30fcd6a1da 100644 --- a/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/SyncOperatorApp.scala +++ b/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/SyncOperatorApp.scala @@ -27,7 +27,6 @@ import org.lfdecentralizedtrust.splice.environment.{ SpliceLedgerClient, } import org.lfdecentralizedtrust.splice.scan.admin.api.client.ScanConnection -import org.lfdecentralizedtrust.splice.store.db.DbAppStore import org.lfdecentralizedtrust.splice.syncoperator.automation.SyncOperatorAutomationService import org.lfdecentralizedtrust.splice.syncoperator.config.SyncOperatorAppBackendConfig import org.lfdecentralizedtrust.splice.syncoperator.metrics.SyncOperatorAppMetrics @@ -112,11 +111,6 @@ class SyncOperatorApp( synchronizerId <- appInitStep("Get the synchronizer id from the sequencer") { servedSynchronizerId(sequencerAdminConnection) } - // Only used to partition the store's ingestion offsets; purchases are ingested regardless - // of it, see SyncOperatorStore.contractFilter. - domainMigrationId <- appInitStep(s"Resolving domain migration id") { - resolveDomainMigrationId(scanConnection) - } storeKey = SyncOperatorStore.Key( operatorParty = partyId, dsoParty = dsoParty, @@ -127,7 +121,9 @@ class SyncOperatorApp( storage, loggerFactory, retryProvider, - domainMigrationId, + // MIGRATION_ID is frozen network-wide and logical synchronizer upgrades carry a serial id + // instead, so the store's partition never has to move. + 0L, participantId, config.automation.ingestion, config.parameters.defaultLimit, @@ -190,26 +186,6 @@ class SyncOperatorApp( logger, ) - private def resolveDomainMigrationId( - scanConnection: ScanConnection - )(implicit traceContext: TraceContext): Future[Long] = - DbAppStore.getHighestKnownMigrationId(storage).flatMap { - case Some(migrationId) => - logger.info(s"Resolved domain migration id $migrationId from the local store offsets") - Future.successful(migrationId) - case None => - retryProvider.getValueWithRetries( - RetryFor.WaitingOnInitDependency, - "sync_operator_domain_migration_id", - s"Wait for domain migration id to be available", - scanConnection.getMigrationId().map { migrationId => - logger.info(s"Resolved domain migration id $migrationId from scan") - migrationId - }, - logger, - ) - } - protected[this] override def automationServices(st: SyncOperatorApp.State) = Seq(st.automation) } From 89d0017efda7bb367d7e82dbddbf4b0929ae5a04 Mon Sep 17 00:00:00 2001 From: sadiq1971 Date: Fri, 28 Aug 2026 14:27:42 +0600 Subject: [PATCH 3/3] feat: grant purchased traffic on the dedicated sequencer [ci] Binds the reusable reconciliation trigger from #15 to the operator's own sequencer and store, so purchases naming its synchronizer are granted there. Enables traffic control on the splitwell synchronizer in the canton bootstrap, since a member only gets a traffic state, and so can only be granted extra traffic, on a synchronizer that has it. splitwellUpgrade is left alone. The integration test registers the synchronizer by DSO vote, buys traffic for it with the registration disclosed, and asserts the operator raises the limit on that sequencer. It calls the buy choice directly because the wallet path does not carry a registration yet. Signed-off-by: sadiq1971 --- .../SyncOperatorTrafficIntegrationTest.scala | 130 ++++++++++++++++++ .../splice/syncoperator/SyncOperatorApp.scala | 2 + ...cileDedicatedSequencerTrafficTrigger.scala | 53 +++++++ .../SyncOperatorAutomationService.scala | 13 ++ .../config/SyncOperatorAppConfig.scala | 2 + bootstrap-canton.sc | 28 +++- test-full-class-names.log | 1 + 7 files changed, 225 insertions(+), 4 deletions(-) create mode 100644 apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/tests/SyncOperatorTrafficIntegrationTest.scala create mode 100644 apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/automation/ReconcileDedicatedSequencerTrafficTrigger.scala diff --git a/apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/tests/SyncOperatorTrafficIntegrationTest.scala b/apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/tests/SyncOperatorTrafficIntegrationTest.scala new file mode 100644 index 0000000000..48d9af0243 --- /dev/null +++ b/apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/tests/SyncOperatorTrafficIntegrationTest.scala @@ -0,0 +1,130 @@ +// Copyright (c) 2024 Digital Asset (Switzerland) GmbH and/or its affiliates. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +package org.lfdecentralizedtrust.splice.integration.tests + +import com.digitalasset.canton.SynchronizerAlias +import com.digitalasset.canton.data.CantonTimestamp +import com.digitalasset.canton.topology.Member +import org.lfdecentralizedtrust.splice.codegen.java.splice +import org.lfdecentralizedtrust.splice.codegen.java.splice.round.IssuingMiningRound +import org.lfdecentralizedtrust.splice.codegen.java.splice.types.Round +import org.lfdecentralizedtrust.splice.integration.EnvironmentDefinition +import org.lfdecentralizedtrust.splice.integration.tests.SpliceTests.{ + IntegrationTest, + SpliceTestConsoleEnvironment, +} +import org.lfdecentralizedtrust.splice.util.{DisclosedContracts, SynchronizerFeesTestUtil, WalletTestUtil} + +import scala.jdk.CollectionConverters.* +import scala.jdk.OptionConverters.* + +/** The buy path for a registered synchronizer is not exposed through the wallet yet + * (ChainSafe/canton-extending-mainnet#39), so this exercises `AmuletRules_BuyMemberTraffic` + * directly with the registration disclosed, which is what the wallet will eventually do. + */ +class SyncOperatorTrafficIntegrationTest + extends IntegrationTest + with SynchronizerFeesTestUtil + with WalletTestUtil { + + private val trafficAmount = 1_000_000L + + override def environmentDefinition: SpliceEnvironmentDefinition = + EnvironmentDefinition + .fromResources( + Seq("simple-topology-1sv.conf", "sync-operator-topology.conf"), + this.getClass.getSimpleName, + ) + .withStandardSetup + + "sync operator" should { + + "grant traffic purchased for its synchronizer on its own sequencer" in { implicit env => + val operatorParty = syncOperatorBackend.appState.store.key.operatorParty + val dsoParty = sv1Backend.getDsoInfo().dsoParty + val dsoRules = sv1Backend.getDsoInfo().dsoRules + // The operator is pointed at the splitwell sequencer, so that is the synchronizer whose + // traffic it grants. Alice's participant is a member of it. + val synchronizerId = aliceValidatorBackend.participantClientWithAdminToken.synchronizers + .id_of(SynchronizerAlias.tryCreate("splitwell")) + .logical + val member = aliceValidatorBackend.participantClient.id + + val registration = clue("the DSO registers the synchronizer to this operator") { + val result = sv1Backend.participantClientWithAdminToken.ledger_api_extensions.commands + .submitWithResult( + sv1Backend.config.ledgerApiUser, + actAs = Seq(dsoParty), + readAs = Seq(dsoParty), + update = dsoRules.contractId.exerciseDsoRules_RegisterSynchronizer( + synchronizerId.toProtoPrimitive, + operatorParty.toProtoPrimitive, + ), + ) + result.exerciseResult.registeredSynchronizerCid + } + + val limitBefore = extraTrafficLimit(member) + + clue("alice buys traffic for the registered synchronizer") { + val aliceParty = onboardWalletUser(aliceWalletClient, aliceValidatorBackend) + aliceWalletClient.tap(walletUsdToAmulet(100.0)) + val transferContext = + sv1ScanBackend.getTransferContextWithInstances(CantonTimestamp.now()) + val amulets = aliceWalletClient.list().amulets.map(_.contract.contractId.contractId) + + aliceValidatorBackend.participantClientWithAdminToken.ledger_api_extensions.commands + .submitWithResult( + aliceValidatorBackend.config.ledgerApiUser, + actAs = Seq(aliceParty), + readAs = Seq(aliceParty), + update = transferContext.amuletRules.contract.contractId + .exerciseAmuletRules_BuyMemberTraffic( + amulets + .map[splice.amuletrules.TransferInput](cid => + new splice.amuletrules.transferinput.InputAmulet( + new splice.amulet.Amulet.ContractId(cid) + ) + ) + .asJava, + new splice.amuletrules.TransferContext( + transferContext.latestOpenMiningRound.contract.contractId, + Map.empty[Round, IssuingMiningRound.ContractId].asJava, + Map.empty[String, splice.amulet.ValidatorRight.ContractId].asJava, + None.toJava, + ), + aliceParty.toProtoPrimitive, + member.toProtoPrimitive, + synchronizerId.toProtoPrimitive, + // a registered synchronizer is pinned to migration id 0 + 0L, + trafficAmount, + Some(dsoParty.toProtoPrimitive).toJava, + Some(registration).toJava, + ), + disclosedContracts = DisclosedContracts + .forTesting( + transferContext.amuletRules, + transferContext.latestOpenMiningRound, + ) + .toLedgerApiDisclosedContracts, + ) + } + + clue("the operator grants it on the splitwell sequencer") { + eventually() { + extraTrafficLimit(member) shouldBe (limitBefore + trafficAmount) + } + } + } + } + + private def extraTrafficLimit( + member: Member + )(implicit env: SpliceTestConsoleEnvironment): Long = + syncOperatorBackend.appState.sequencerAdminConnection + .lookupSequencerTrafficControlState(member) + .futureValue + .fold(0L)(_.extraTrafficLimit.value) +} diff --git a/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/SyncOperatorApp.scala b/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/SyncOperatorApp.scala index 30fcd6a1da..490e712012 100644 --- a/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/SyncOperatorApp.scala +++ b/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/SyncOperatorApp.scala @@ -149,6 +149,8 @@ class SyncOperatorApp( ledgerClient, retryProvider, config.parameters, + sequencerAdminConnection, + config.trafficBalanceReconciliationDelay, loggerFactory, packageVersionSupport, ) diff --git a/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/automation/ReconcileDedicatedSequencerTrafficTrigger.scala b/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/automation/ReconcileDedicatedSequencerTrafficTrigger.scala new file mode 100644 index 0000000000..65d1e47967 --- /dev/null +++ b/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/automation/ReconcileDedicatedSequencerTrafficTrigger.scala @@ -0,0 +1,53 @@ +// Copyright (c) 2024 Digital Asset (Switzerland) GmbH and/or its affiliates. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +package org.lfdecentralizedtrust.splice.syncoperator.automation + +import com.digitalasset.canton.config.NonNegativeFiniteDuration +import com.digitalasset.canton.topology.Member +import com.digitalasset.canton.tracing.TraceContext +import io.opentelemetry.api.trace.Tracer +import org.apache.pekko.stream.Materializer +import org.lfdecentralizedtrust.splice.automation.{ + ReconcileSequencerLimitWithMemberTrafficTriggerBase, + TriggerContext, +} +import org.lfdecentralizedtrust.splice.environment.SequencerAdminConnection +import org.lfdecentralizedtrust.splice.syncoperator.store.SyncOperatorStore + +import scala.concurrent.{ExecutionContext, Future} + +/** Reconciles the traffic purchased for this operator's synchronizer with its sequencer. The store + * only holds purchases naming that synchronizer, so the base trigger's per-contract check is a + * safeguard rather than the filter. + */ +class ReconcileDedicatedSequencerTrafficTrigger( + override protected val context: TriggerContext, + store: SyncOperatorStore, + sequencerConnection: SequencerAdminConnection, + trafficBalanceReconciliationDelay: NonNegativeFiniteDuration, +)(implicit + ec: ExecutionContext, + mat: Materializer, + tracer: Tracer, +) extends ReconcileSequencerLimitWithMemberTrafficTriggerBase( + store, + trafficBalanceReconciliationDelay, + ) { + + override protected def sequencerAdminConnection()(implicit + tc: TraceContext + ): Future[SequencerAdminConnection] = + Future.successful(sequencerConnection) + + override protected def getTotalPurchasedMemberTraffic(memberId: Member)(implicit + tc: TraceContext + ): Future[Long] = + store.getTotalPurchasedMemberTraffic(memberId) + + override protected def trafficLimitOffset(memberId: Member)(implicit + tc: TraceContext + ): Future[Either[String, Long]] = + // No prior consumption to carry. + Future.successful(Right(0L)) +} diff --git a/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/automation/SyncOperatorAutomationService.scala b/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/automation/SyncOperatorAutomationService.scala index 3821e1e148..3b7054d88f 100644 --- a/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/automation/SyncOperatorAutomationService.scala +++ b/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/automation/SyncOperatorAutomationService.scala @@ -3,6 +3,7 @@ package org.lfdecentralizedtrust.splice.syncoperator.automation +import com.digitalasset.canton.config.NonNegativeFiniteDuration import com.digitalasset.canton.logging.NamedLoggerFactory import com.digitalasset.canton.resource.DbStorage import com.digitalasset.canton.time.Clock @@ -17,6 +18,7 @@ import org.lfdecentralizedtrust.splice.config.{AutomationConfig, SpliceParameter import org.lfdecentralizedtrust.splice.environment.{ PackageVersionSupport, RetryProvider, + SequencerAdminConnection, SpliceLedgerClient, } import org.lfdecentralizedtrust.splice.store.DomainTimeSynchronization @@ -33,6 +35,8 @@ class SyncOperatorAutomationService( ledgerClient: SpliceLedgerClient, retryProvider: RetryProvider, params: SpliceParametersConfig, + sequencerConnection: SequencerAdminConnection, + trafficBalanceReconciliationDelay: NonNegativeFiniteDuration, protected val loggerFactory: NamedLoggerFactory, packageVersionSupport: PackageVersionSupport, )(implicit @@ -59,6 +63,15 @@ class SyncOperatorAutomationService( triggerContext, ) ) + + registerTrigger( + new ReconcileDedicatedSequencerTrafficTrigger( + triggerContext, + store, + sequencerConnection, + trafficBalanceReconciliationDelay, + ) + ) } object SyncOperatorAutomationService extends AutomationServiceCompanion { diff --git a/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/config/SyncOperatorAppConfig.scala b/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/config/SyncOperatorAppConfig.scala index 0204f28a27..606dc8210d 100644 --- a/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/config/SyncOperatorAppConfig.scala +++ b/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/config/SyncOperatorAppConfig.scala @@ -31,6 +31,8 @@ case class SyncOperatorAppBackendConfig( sequencer: SyncOperatorSequencerConfig, override val automation: AutomationConfig = AutomationConfig(), parameters: SpliceParametersConfig = SpliceParametersConfig(batching = BatchingConfig()), + trafficBalanceReconciliationDelay: NonNegativeFiniteDuration = + NonNegativeFiniteDuration.ofSeconds(10), // Set to false to disable the DB-level exclusive lock that prevents two sync operator instances // from running concurrently against the same database. Only disable for migration scenarios // where intentional overlap is required. diff --git a/bootstrap-canton.sc b/bootstrap-canton.sc index 4da9344cd7..06e7c37bd9 100644 --- a/bootstrap-canton.sc +++ b/bootstrap-canton.sc @@ -13,6 +13,9 @@ import com.digitalasset.canton.console.{ import com.digitalasset.canton.SynchronizerAlias import com.digitalasset.canton.synchronizer.config.SynchronizerParametersConfig import com.digitalasset.canton.protocol.DynamicSynchronizerParameters +import com.digitalasset.canton.admin.api.client.data.TrafficControlParameters +import com.digitalasset.canton.config.PositiveFiniteDuration +import com.digitalasset.canton.config.RequireTypes.NonNegativeLong import com.digitalasset.canton.topology.transaction.SignedTopologyTransaction.GenericSignedTopologyTransaction import com.digitalasset.canton.topology.transaction.TopologyChangeOp import com.digitalasset.canton.version.ProtocolVersion @@ -36,10 +39,25 @@ def staticParameters(sequencer: LocalInstanceReference) = .map(StaticSynchronizerParameters(_)) .getOrElse(sys.error("whatever")) +// Canton's own defaults. A member only gets a traffic state, and so can only be granted extra +// traffic, on a synchronizer that has traffic control enabled. +val defaultTrafficControlParameters = TrafficControlParameters( + maxBaseTrafficAmount = NonNegativeLong.tryCreate(10 * 20 * 1024), + readVsWriteScalingFactor = PositiveInt.tryCreate(200), + maxBaseTrafficAccumulationDuration = PositiveFiniteDuration.ofMinutes(10), + setBalanceRequestSubmissionWindowSize = PositiveFiniteDuration.ofMinutes(5), + enforceRateLimiting = true, + baseEventCost = NonNegativeLong.zero, + freeConfirmationResponses = false, +) + def bootstrapOtherDomain( name: String, sequencer: LocalSequencerReference, mediator: LocalMediatorReference, + // Traffic control is off by default here: only synchronizers that stand in for a dedicated one + // need it, and enabling it everywhere would change what every other test sequences against. + enableTrafficControl: Boolean = false, ) = { bootstrap.synchronizer( name, @@ -67,6 +85,8 @@ def bootstrapOtherDomain( ), preparationTimeRecordTimeTolerance = NonNegativeFiniteDuration.ofHours(24), mediatorDeduplicationTimeout = NonNegativeFiniteDuration.ofHours(48), + trafficControl = + if (enableTrafficControl) Some(defaultTrafficControlParameters) else parameters.trafficControl, ), signedBy = Some(sequencer.id.uid.namespace.fingerprint), // This is test code so just force the change. @@ -74,10 +94,10 @@ def bootstrapOtherDomain( ) } -Seq( - ("splitwell", splitwellSequencer, splitwellMediator), - ("splitwellUpgrade", splitwellUpgradeSequencer, splitwellUpgradeMediator), -).foreach((bootstrapOtherDomain _).tupled) +// splitwell is the only non-global synchronizer a sync operator can be pointed at today, so it +// carries traffic control; see apps/app/src/test/resources/sync-operator-topology.conf. +bootstrapOtherDomain("splitwell", splitwellSequencer, splitwellMediator, enableTrafficControl = true) +bootstrapOtherDomain("splitwellUpgrade", splitwellUpgradeSequencer, splitwellUpgradeMediator) // These user allocations are only there // for local testing. Our tests allocate their own users. diff --git a/test-full-class-names.log b/test-full-class-names.log index 88aa72960e..22b2a37942 100644 --- a/test-full-class-names.log +++ b/test-full-class-names.log @@ -54,6 +54,7 @@ org.lfdecentralizedtrust.splice.integration.tests.SvReconcileBftSequencingParame org.lfdecentralizedtrust.splice.integration.tests.SvReconcileSynchronizerConfigIntegrationTest org.lfdecentralizedtrust.splice.integration.tests.SvStateManagementIntegrationTest org.lfdecentralizedtrust.splice.integration.tests.SyncOperatorIntegrationTest +org.lfdecentralizedtrust.splice.integration.tests.SyncOperatorTrafficIntegrationTest org.lfdecentralizedtrust.splice.integration.tests.TestTokenV2SettlementIntegrationTest org.lfdecentralizedtrust.splice.integration.tests.TokenStandardAllocationIntegrationTest org.lfdecentralizedtrust.splice.integration.tests.TokenStandardCliIntegrationTest