feat: grant purchased traffic on the dedicated sequencer

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)
+}

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)
 }

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))
+}

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 {

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.

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,17 +85,19 @@ 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.
     force = ForceFlags(ForceFlag.PreparationTimeRecordTimeToleranceIncrease),
   )
 }
 
-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.

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