test: sync operator integration test and CI harness

.github/workflows/build.yml

@@ -150,6 +150,19 @@ jobs:
       protocol_version: ${{ inputs.protocol_version }}
     secrets: inherit
 
+  scala_test_sync_operator:
+    uses: ./.github/workflows/build.scala_test.yml
+    with:
+      runs_on: self-hosted-k8s-medium
+      test_names_file: 'test-full-class-names-sync-operator.log'
+      start_canton_options: -w -t
+      parallelism: 1
+      test_name: sync-operator
+      commit_sha: ${{ inputs.commit_sha }}
+      daml_base_version: ${{ inputs.daml_base_version }}
+      protocol_version: ${{ inputs.protocol_version }}
+    secrets: inherit
+
   scala_test_logical_sync_upgrade:
     uses: ./.github/workflows/build.scala_test.yml
     with:
@@ -343,6 +356,7 @@ jobs:
     - scala_test_resource_intensive
     - scala_test_key_download
     - scala_test_record_time_tolerance
+    - scala_test_sync_operator
     - scala_test_logical_sync_upgrade
     - scala_test_roll_forward_lsu
     - scala_test_signatures

apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/EnvironmentDefinition.scala

@@ -502,6 +502,30 @@ case class EnvironmentDefinition(
       )
   }
 
+  /** Drops the splitwell apps and leaves alice's validator as the only one connected to the
+    * splitwell synchronizer.
+    */
+  def withOnlyAliceValidatorConnectingToSplitwell: EnvironmentDefinition =
+    withValidatorsDisconnectedFromSplitwell("bobValidator", "splitwellValidator")
+
+  /** Drops all splitwell apps, and clears all of the given validators' extra synchronizers, not
+    * just splitwell, so the global synchronizer is the only one they connect to.
+    */
+  def withValidatorsDisconnectedFromSplitwell(validators: String*): EnvironmentDefinition =
+    this
+      .addConfigTransform((_, conf) =>
+        conf.copy(
+          splitwellApps = Map.empty,
+          validatorApps = validators.foldLeft(conf.validatorApps) { (apps, name) =>
+            apps.updatedWith(InstanceName.tryCreate(name)) {
+              _.map { validatorConfig =>
+                validatorConfig.copy(domains = validatorConfig.domains.copy(extra = Seq.empty))
+              }
+            }
+          },
+        )
+      )
+
   def clearConfigTransforms(): EnvironmentDefinition =
     copy(configTransformsWithContext = _ => Seq())
 

apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/plugins/WaitForPorts.scala

@@ -28,6 +28,7 @@ case class WaitForPorts(extraPortsToWaitFor: Seq[(String, Int)])
     config.svApps.foreach(sv => waitForPort(sv._1, sv._2.adminApi.port.unwrap))
     config.splitwellApps.foreach(sw => waitForPort(sw._1, sw._2.adminApi.port.unwrap))
     config.scanApps.foreach(scan => waitForPort(scan._1, scan._2.adminApi.port.unwrap))
+    config.syncOperatorApps.foreach(so => waitForPort(so._1, so._2.adminApi.port.unwrap))
     extraPortsToWaitFor.foreach(p => waitForPort(InstanceName.tryCreate(p._1), p._2))
     // Wait long enough that the prometheus server can start.
     config.monitoring.metrics.reporters.foreach {

apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/tests/SyncOperatorIntegrationTest.scala

@@ -0,0 +1,40 @@
+// 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 org.lfdecentralizedtrust.splice.integration.EnvironmentDefinition
+import org.lfdecentralizedtrust.splice.integration.tests.SpliceTests.IntegrationTest
+
+class SyncOperatorIntegrationTest extends IntegrationTest {
+
+  override def environmentDefinition: SpliceEnvironmentDefinition =
+    EnvironmentDefinition
+      .fromResources(
+        Seq("simple-topology-1sv.conf", "sync-operator-topology.conf"),
+        this.getClass.getSimpleName,
+      )
+      .withOnlyAliceValidatorConnectingToSplitwell
+      .withStandardSetup
+
+  "sync operator" should {
+    "restart cleanly" in { implicit env =>
+      syncOperatorBackend.stop()
+      syncOperatorBackend.startSync()
+    }
+
+    "report liveness and readiness" in { implicit env =>
+      syncOperatorBackend.httpLive shouldBe true
+      syncOperatorBackend.httpReady shouldBe true
+    }
+
+    "take its synchronizer id from the sequencer it is configured with" in { implicit env =>
+      // Alice's participant is the one still connected to splitwell in this topology.
+      val served = aliceValidatorBackend.participantClientWithAdminToken.synchronizers
+        .id_of(SynchronizerAlias.tryCreate("splitwell"))
+        .logical
+      syncOperatorBackend.appState.synchronizerId shouldBe served
+    }
+  }
+}

apps/app/src/test/scala/org/lfdecentralizedtrust/splice/util/CommonAppInstanceReferences.scala

@@ -10,6 +10,7 @@ import org.lfdecentralizedtrust.splice.console.{
   SplitwellAppClientReference,
   SvAppBackendReference,
   SvAppClientReference,
+  SyncOperatorAppBackendReference,
   ValidatorAppBackendReference,
   ValidatorAppClientReference,
   WalletAppClientReference,
@@ -297,6 +298,12 @@ trait CommonAppInstanceReferences {
     "providerSplitwellBackend"
   )
 
+  def syncOperatorBackend(implicit
+      env: SpliceTestConsoleEnvironment
+  ): SyncOperatorAppBackendReference = syncop(
+    "syncOperator"
+  )
+
   def svb(name: String)(implicit env: SpliceTestConsoleEnvironment): SvAppBackendReference =
     env.svs.local
       .find(_.name == name)
@@ -348,6 +355,13 @@ trait CommonAppInstanceReferences {
       .find(_.name == name)
       .getOrElse(sys.error(s"remote splitwell [$name] not configured"))
 
+  def syncop(
+      name: String
+  )(implicit env: SpliceTestConsoleEnvironment): SyncOperatorAppBackendReference =
+    env.syncOperators.local
+      .find(_.name == name)
+      .getOrElse(sys.error(s"local sync operator [$name] not configured"))
+
   def scanb(
       name: String
   )(implicit env: SpliceTestConsoleEnvironment): ScanAppBackendReference =

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,32 @@ def staticParameters(sequencer: LocalInstanceReference) =
     .map(StaticSynchronizerParameters(_))
     .getOrElse(sys.error("whatever"))
 
+// The base rate is mainnet's unless start-canton.sh sets it; -t sets it to zero for the sync operator tests.
+val splitwellMaxBaseTrafficAmount = Option(System.getenv("SPLITWELL_MAX_BASE_TRAFFIC_AMOUNT"))
+  .filter(_.nonEmpty)
+  .map(amount => NonNegativeLong.tryCreate(amount.toLong))
+  .getOrElse(NonNegativeLong.tryCreate(200000))
+
+// The global synchronizer's parameters on mainnet, apart from the base rate. 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 = splitwellMaxBaseTrafficAmount,
+  readVsWriteScalingFactor = PositiveInt.tryCreate(4),
+  maxBaseTrafficAccumulationDuration = PositiveFiniteDuration.ofMinutes(20),
+  setBalanceRequestSubmissionWindowSize = PositiveFiniteDuration.ofMinutes(2),
+  enforceRateLimiting = true,
+  baseEventCost = NonNegativeLong.zero,
+  freeConfirmationResponses = true,
+)
+
 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 +92,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.

build.sbt

@@ -2575,6 +2575,10 @@ updateTestConfigForParallelRuns := {
     name contains "LsuIntegrationTest"
   def isLSURollForwardTest(name: String): Boolean =
     name contains "RollForwardLsu"
+  // These run in their own CI job with a zero base traffic rate on the splitwell synchronizer;
+  // see start-canton.sh -t.
+  def isSyncOperatorTest(name: String): Boolean =
+    name contains "SyncOperator"
 
   val allTestNames =
     definedTests
@@ -2687,6 +2691,11 @@ updateTestConfigForParallelRuns := {
       "test-full-class-names-lsu.log",
       (t: String) => isLSUTest(t),
     ),
+    (
+      "sync operator tests",
+      "test-full-class-names-sync-operator.log",
+      (t: String) => isSyncOperatorTest(t),
+    ),
     (
       "tests with wall clock time",
       "test-full-class-names.log",

project/ignore-patterns/canton-standalone-sync-operator.ignore.txt

@@ -0,0 +1,5 @@
+# With a zero base rate, a member's sends are refused until the sync operator grants it traffic.
+# The participant retries them automatically. Every layer that surfaces the refusal carries
+# the error code; the outbox retry line carries the transaction hashes instead.
+SEQUENCER_NOT_ENOUGH_TRAFFIC_CREDIT
+synchronizer outbox flusher.*splitwell.*failed the following topology transactions

start-canton.sh

@@ -18,6 +18,7 @@ function usage() {
   echo "  -F               same as -f, but does not start cometBFT, but rather assumes it is already running"
   echo "  -e               start canton using the canton provided BFT sequencer"
   echo "  -m               collect metrics and send them to our CI prometheus instance"
+  echo "  -t               zero base traffic rate on the splitwell synchronizer"
   echo "  -c <canton>      start a custom canton binary instead of the one on the PATH"
   echo "  -B <script>      path to a custom canton bootstrap script"
   echo "  -k               allow downloading private keys via the admin API (needed, e.g., for identities dumps)"
@@ -36,6 +37,8 @@ use_bft=0
 collect_metrics=0
 enable_key_download=0
 logFileHint=canton
+# Base traffic rate on the splitwell synchronizer in bytes; empty means mainnet's.
+splitwell_max_base_traffic_amount="${SPLITWELL_MAX_BASE_TRAFFIC_AMOUNT-}"
 
 args=$(getopt -o "hdDap:cB:wsbtfFegmk" -l "help" -- "$@")
 
@@ -99,6 +102,11 @@ do
             enable_key_download=1
             echo "allowing downloads of private keys"
             ;;
+        -t)
+            splitwell_max_base_traffic_amount=0
+            logFileHint=canton-standalone-sync-operator
+            echo "starting canton with a zero base traffic rate on the splitwell synchronizer"
+            ;;
         -B)
             bootstrapScriptPath="$2"
             logFileHint=canton-missing-signatures
@@ -228,6 +236,7 @@ tmux_cmd_canton() {
     "EXTRA_CLASSPATH=$COMETBFT_DRIVER/driver.jar \
      COMETBFT_DOCKER_IP=${COMETBFT_DOCKER_IP-} \
      LOG_LEVEL_API_REQUEST=DEBUG \
+     SPLITWELL_MAX_BASE_TRAFFIC_AMOUNT=${splitwell_max_base_traffic_amount-} \
      CANTON_PROTOCOL_VERSION=${CANTON_PROTOCOL_VERSION:-37} CANTON_TOKEN_FILENAME=$tokensFile CANTON_PARTICIPANTS_FILENAME=$participantsFile JAVA_TOOL_OPTIONS=\"$JAVA_TOOL_OPTIONS\" $CANTON \
       -c $baseConfig $confOverrides \
       --log-level-canton=DEBUG \

test-full-class-names-sync-operator.log

@@ -0,0 +1 @@
+org.lfdecentralizedtrust.splice.integration.tests.SyncOperatorIntegrationTest