feat: let the sync operator upgrade its dedicated synchronizer

.gitignore

@@ -85,3 +85,7 @@ lnav*work
 
 # VS Code
 .vscode/
+
+# LSU schedule written by LocalNetDedicatedSyncLsuIntegrationTest
+cluster/compose/localnet/conf/splice/sync-operator/lsu-topology-freeze-time
+cluster/compose/localnet/conf/splice/sync-operator/lsu-upgrade-time

apps/app/src/main/scala/org/lfdecentralizedtrust/splice/config/ConfigTransforms.scala

@@ -24,7 +24,10 @@ import org.lfdecentralizedtrust.splice.sv.automation.singlesv.offboarding.{
 }
 import org.lfdecentralizedtrust.splice.sv.config.*
 import org.lfdecentralizedtrust.splice.sv.SvAppClientConfig
-import org.lfdecentralizedtrust.splice.syncoperator.config.SyncOperatorAppBackendConfig
+import org.lfdecentralizedtrust.splice.syncoperator.config.{
+  SyncOperatorAppBackendConfig,
+  SyncOperatorSynchronizerNodeConfig,
+}
 import org.lfdecentralizedtrust.splice.validator.config.{
   AnsAppExternalClientConfig,
   ValidatorAppBackendConfig,
@@ -558,8 +561,10 @@ object ConfigTransforms {
           conf
             .focus(_.participantClient)
             .modify(portTransform(bump, _))
-            .focus(_.sequencer.adminApi)
+            .focus(_.synchronizerNodes.current)
             .modify(portTransform(bump, _))
+            .focus(_.synchronizerNodes.successor)
+            .modify(_.map(portTransform(bump, _)))
         else conf
       ),
     )
@@ -914,6 +919,19 @@ object ConfigTransforms {
   private def portTransform(bump: Int, c: SvMediatorConfig): SvMediatorConfig =
     c.focus(_.adminApi).modify(portTransform(bump, _))
 
+  private def portTransform(
+      bump: Int,
+      c: SyncOperatorSynchronizerNodeConfig,
+  ): SyncOperatorSynchronizerNodeConfig =
+    c.focus(_.sequencer.adminApi)
+      .modify(portTransform(bump, _))
+      .focus(_.sequencer.internalApi)
+      .modify(_.map(portTransform(bump, _)))
+      .focus(_.sequencer.externalPublicApiUrl)
+      .modify(_.map(bumpUrl(bump, _)))
+      .focus(_.mediator)
+      .modify(_.map(_.focus(_.adminApi).modify(portTransform(bump, _))))
+
   private def portTransform(
       bump: Int,
       c: SvSynchronizerNodeConfig,

apps/app/src/main/scala/org/lfdecentralizedtrust/splice/config/SpliceConfig.scala

@@ -36,7 +36,11 @@ import org.lfdecentralizedtrust.splice.sv.config.SvOnboardingConfig.FoundDso
 import org.lfdecentralizedtrust.splice.syncoperator.config.{
   SyncOperatorAppBackendConfig,
   SyncOperatorAppClientConfig,
+  SyncOperatorLsuConfig,
+  SyncOperatorMediatorConfig,
   SyncOperatorSequencerConfig,
+  SyncOperatorSynchronizerNodeConfig,
+  SyncOperatorSynchronizerNodesConfig,
 }
 import org.lfdecentralizedtrust.splice.util.{
   Codec,
@@ -998,6 +1002,16 @@ object SpliceConfig {
       deriveReader[SplitwellAppClientConfig]
     implicit val syncOperatorSequencerConfigReader: ConfigReader[SyncOperatorSequencerConfig] =
       deriveReader[SyncOperatorSequencerConfig]
+    implicit val syncOperatorMediatorConfigReader: ConfigReader[SyncOperatorMediatorConfig] =
+      deriveReader[SyncOperatorMediatorConfig]
+    implicit val syncOperatorSynchronizerNodeConfigReader
+        : ConfigReader[SyncOperatorSynchronizerNodeConfig] =
+      deriveReader[SyncOperatorSynchronizerNodeConfig]
+    implicit val syncOperatorSynchronizerNodesConfigReader
+        : ConfigReader[SyncOperatorSynchronizerNodesConfig] =
+      deriveReader[SyncOperatorSynchronizerNodesConfig]
+    implicit val syncOperatorLsuConfigReader: ConfigReader[SyncOperatorLsuConfig] =
+      deriveReader[SyncOperatorLsuConfig]
     implicit val syncOperatorConfigReader: ConfigReader[SyncOperatorAppBackendConfig] =
       deriveReader[SyncOperatorAppBackendConfig]
     implicit val syncOperatorClientConfigReader: ConfigReader[SyncOperatorAppClientConfig] =
@@ -1330,6 +1344,16 @@ object SpliceConfig {
       deriveWriter[SplitwellAppClientConfig]
     implicit val syncOperatorSequencerConfigWriter: ConfigWriter[SyncOperatorSequencerConfig] =
       deriveWriter[SyncOperatorSequencerConfig]
+    implicit val syncOperatorMediatorConfigWriter: ConfigWriter[SyncOperatorMediatorConfig] =
+      deriveWriter[SyncOperatorMediatorConfig]
+    implicit val syncOperatorSynchronizerNodeConfigWriter
+        : ConfigWriter[SyncOperatorSynchronizerNodeConfig] =
+      deriveWriter[SyncOperatorSynchronizerNodeConfig]
+    implicit val syncOperatorSynchronizerNodesConfigWriter
+        : ConfigWriter[SyncOperatorSynchronizerNodesConfig] =
+      deriveWriter[SyncOperatorSynchronizerNodesConfig]
+    implicit val syncOperatorLsuConfigWriter: ConfigWriter[SyncOperatorLsuConfig] =
+      deriveWriter[SyncOperatorLsuConfig]
     implicit val syncOperatorConfigWriter: ConfigWriter[SyncOperatorAppBackendConfig] =
       deriveWriter[SyncOperatorAppBackendConfig]
     implicit val syncOperatorClientConfigWriter: ConfigWriter[SyncOperatorAppClientConfig] =

apps/app/src/test/resources/sync-operator-topology.conf

@@ -24,7 +24,7 @@ canton {
       admin-api.port = 5116
       participant-client = ${canton.validator-apps.splitwellValidator.participant-client}
       operator-user = ${canton.validator-apps.splitwellValidator.app-instances.syncOperator.service-user}
-      sequencer.admin-api.port = 5709
+      synchronizer-nodes.current.sequencer.admin-api.port = 5709
     }
   }
 }

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

@@ -2,99 +2,14 @@ package org.lfdecentralizedtrust.splice.integration.tests
 
 import com.digitalasset.canton.admin.api.client.data.OnboardingRestriction
 import com.digitalasset.canton.config.RequireTypes.NonNegativeLong
-import com.digitalasset.canton.topology.{PartyId, SynchronizerId}
-import monocle.Monocle.toAppliedFocusOps
-import org.lfdecentralizedtrust.splice.auth.AuthUtil
-import org.lfdecentralizedtrust.splice.codegen.java.da.time.types.RelTime
-import org.lfdecentralizedtrust.splice.codegen.java.splice
-import org.lfdecentralizedtrust.splice.codegen.java.splice.decentralizedsynchronizer.RegisteredSynchronizer
-import org.lfdecentralizedtrust.splice.codegen.java.splice.decentralizedsynchronizer.GovernanceParameters
-import org.lfdecentralizedtrust.splice.codegen.java.splice.dsorules.DsoRules_RegisterSynchronizer
-import org.lfdecentralizedtrust.splice.codegen.java.splice.dsorules.actionrequiringconfirmation.ARC_DsoRules
-import org.lfdecentralizedtrust.splice.codegen.java.splice.dsorules.dsorules_actionrequiringconfirmation.SRARC_RegisterSynchronizer
-import org.lfdecentralizedtrust.splice.codegen.java.splice.round.IssuingMiningRound
-import org.lfdecentralizedtrust.splice.codegen.java.splice.types.Round
-import org.lfdecentralizedtrust.splice.console.ParticipantClientReference
-import org.lfdecentralizedtrust.splice.integration.EnvironmentDefinition
-import org.lfdecentralizedtrust.splice.integration.tests.SpliceTests.IntegrationTestWithIsolatedEnvironment
-import org.lfdecentralizedtrust.splice.util.{ContractWithState, DisclosedContracts}
-
-import java.time.Duration
-import scala.concurrent.duration.*
-import scala.jdk.CollectionConverters.*
-import scala.jdk.OptionConverters.*
-import scala.sys.process.*
 
 /** Verifies that the sync operator serves the app-synchronizer as a dedicated synchronizer: it is
   * bootstrapped with a zero base rate and admits only permissioned participants, the DSO registers
   * it to the operator, and a member transacts on it only against traffic it has bought.
   *
   * This spins up the docker-compose localnet with the sync operator enabled (-O)
   */
-class LocalNetDedicatedSyncIntegrationTest extends IntegrationTestWithIsolatedEnvironment {
-
-  override def environmentDefinition: SpliceEnvironmentDefinition =
-    EnvironmentDefinition
-      .fromResources(Seq("localnet-dedicated-sync-topology.conf"), this.getClass.getSimpleName)
-      .updateTestingConfig(
-        _.focus(_.participantsWithoutLapiVerification).replace(Set("app-provider"))
-      )
-      .withManualStart
-
-  // These do nothing as the clients will not actually be connected to the compose setup.
-  override protected def runTokenStandardCliSanityCheck: Boolean = false
-  override lazy val resetRequiredTopologyState = false
-
-  // The user all localnet nodes use for their ledger API access, see
-  // cluster/compose/localnet/env/*-auth-on.env
-  private val ledgerApiUserId = "ledger-api-user"
-
-  private val token = AuthUtil.testToken(AuthUtil.testAudience, ledgerApiUserId, "unsafe")
-
-  // Above the minimum top-up, and enough that a ping visibly draws it down.
-  private val purchasedTraffic = 2_000_000L
-
-  // Covers the traffic fee with a wide margin, the fee itself depends on the amulet price.
-  private val tapAmount = 100_000.0
-
-  // LocalNet runs the apps at their default polling interval, so a change takes a few rounds of
-  // automation to land.
-  private val automationTimeout = 2.minutes
-
-  private def withLocalNet(
-      additionalArgs: Seq[String]
-  )(f: FixtureParam => Any)(implicit env: FixtureParam): Unit =
-    try {
-      val ret = (Seq("build-tools/splice-localnet-compose.sh", "start") ++ additionalArgs).!
-      if (ret != 0) {
-        fail("Failed to start docker-compose localnet with the sync operator")
-      }
-      f(env)
-    } finally {
-      (Seq("build-tools/splice-localnet-compose.sh", "stop", "-D") ++ additionalArgs).!
-    }
-
-  private def participantClient(name: String)(implicit env: FixtureParam) = {
-    val remoteParticipant =
-      env.participants.remote
-        .find(_.name == name)
-        .getOrElse(fail(s"$name participant not found"))
-    new ParticipantClientReference(
-      env,
-      remoteParticipant.name,
-      remoteParticipant.config.copy(token = Some(token)),
-    )
-  }
-
-  private def synchronizerId(
-      participant: ParticipantClientReference,
-      alias: String,
-  ): SynchronizerId =
-    participant.synchronizers
-      .list_connected()
-      .find(_.synchronizerAlias.unwrap == alias)
-      .getOrElse(fail(s"${participant.name} is not connected to $alias"))
-      .synchronizerId
+class LocalNetDedicatedSyncIntegrationTest extends LocalNetDedicatedSyncIntegrationTestBase {
 
   "the sync operator serves the app-synchronizer as a dedicated synchronizer" in { implicit env =>
     withLocalNet(Seq("-O")) { implicit env =>
@@ -119,42 +34,7 @@ class LocalNetDedicatedSyncIntegrationTest extends IntegrationTestWithIsolatedEn
           .find(appSynchronizerId, participant.id) should not be empty
       }
 
-      // The operator acts as the primary party of its ledger API user, onboarded by the
-      // app-provider validator.
-      val operatorParty = eventuallySucceeds(automationTimeout) {
-        participant.ledger_api.users.get("sync-operator").primaryParty.value
-      }
-
-      val sv = sv_client("svClient").copy(token = Some(token))
-      // The SV serves its DSO info only once it is onboarded, which trails the compose start.
-      val svParty = eventuallySucceeds(automationTimeout)(sv.getDsoInfo().svParty)
-
-      clue("the DSO registers the synchronizer to the operator") {
-        sv.createVoteRequest(
-          svParty.toProtoPrimitive,
-          new ARC_DsoRules(
-            new SRARC_RegisterSynchronizer(
-              new DsoRules_RegisterSynchronizer(
-                appSynchronizerId.toProtoPrimitive,
-                operatorParty.toProtoPrimitive,
-                new GovernanceParameters(java.math.BigDecimal.ONE.setScale(10)),
-              )
-            )
-          ),
-          "https://localnet.example/dedicated-sync",
-          "Register the app-synchronizer as a dedicated synchronizer",
-          new RelTime(Duration.ofDays(1).toMillis * 1000),
-          None,
-        )
-      }
-
-      // Scan serves the registration once the vote has closed and Scan itself is up, which
-      // trails the compose start.
-      val registration = eventuallySucceeds(automationTimeout) {
-        scancl("scanClient")
-          .lookupSynchronizerRegistration(appSynchronizerId.toProtoPrimitive)
-          .value
-      }
+      val registration = registerSynchronizer(appSynchronizerId, operatorParty(participant))
 
       clue("with a zero base rate the participant has no allowance of its own") {
         eventually(automationTimeout) {
@@ -190,62 +70,4 @@ class LocalNetDedicatedSyncIntegrationTest extends IntegrationTestWithIsolatedEn
       }
     }
   }
-
-  private def buyTraffic(
-      participant: ParticipantClientReference,
-      buyer: PartyId,
-      synchronizerId: SynchronizerId,
-      registration: ContractWithState[RegisteredSynchronizer.ContractId, RegisteredSynchronizer],
-  )(implicit env: FixtureParam): Unit = {
-    val scan = scancl("scanClient")
-    val transferContext = scan.getTransferContextWithInstances(env.environment.clock.now)
-    val amuletRules = transferContext.amuletRules
-    val openMiningRound = transferContext.latestOpenMiningRound
-
-    val amulet = participant.ledger_api_extensions.commands
-      .submitWithResult(
-        ledgerApiUserId,
-        actAs = Seq(buyer),
-        readAs = Seq(buyer),
-        update = amuletRules.contract.contractId.exerciseAmuletRules_DevNet_Tap(
-          buyer.toProtoPrimitive,
-          BigDecimal(tapAmount).bigDecimal,
-          openMiningRound.contract.contractId,
-        ),
-        disclosedContracts = DisclosedContracts
-          .forTesting(amuletRules, openMiningRound)
-          .toLedgerApiDisclosedContracts,
-      )
-      .exerciseResult
-      .amuletSum
-      .amulet
-
-    val _ = participant.ledger_api_extensions.commands.submitWithResult(
-      ledgerApiUserId,
-      actAs = Seq(buyer),
-      readAs = Seq(buyer),
-      update = amuletRules.contract.contractId.exerciseAmuletRules_BuyMemberTraffic(
-        Seq[splice.amuletrules.TransferInput](
-          new splice.amuletrules.transferinput.InputAmulet(amulet)
-        ).asJava,
-        new splice.amuletrules.TransferContext(
-          openMiningRound.contract.contractId,
-          Map.empty[Round, IssuingMiningRound.ContractId].asJava,
-          Map.empty[String, splice.amulet.ValidatorRight.ContractId].asJava,
-          None.toJava,
-        ),
-        buyer.toProtoPrimitive,
-        participant.id.toProtoPrimitive,
-        synchronizerId.toProtoPrimitive,
-        // a registered synchronizer is pinned to migration id 0
-        0L,
-        purchasedTraffic,
-        Some(scan.getDsoPartyId().toProtoPrimitive).toJava,
-        Some(registration.contractId).toJava,
-      ),
-      disclosedContracts = DisclosedContracts
-        .forTesting(amuletRules, openMiningRound, registration)
-        .toLedgerApiDisclosedContracts,
-    )
-  }
 }