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 2478c861ad..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 @@ -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,12 +121,14 @@ 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, ) - globalSynchronizerId <- appInitStep("Get the decentralized synchronizer id") { + globalSynchronizerId <- appInitStep("Get the global synchronizer id") { scanConnection.getAmuletRulesDomain()(traceContext) } readOnlyLedgerConnection = ledgerClient @@ -153,6 +149,8 @@ class SyncOperatorApp( ledgerClient, retryProvider, config.parameters, + sequencerAdminConnection, + config.trafficBalanceReconciliationDelay, loggerFactory, packageVersionSupport, ) @@ -190,26 +188,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) } 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 e606bac1f3..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 @@ -42,7 +46,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, @@ -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