diff --git a/apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/tests/LsuIntegrationTest.scala b/apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/tests/LsuIntegrationTest.scala index 5da11f8df8..27f7caf82d 100644 --- a/apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/tests/LsuIntegrationTest.scala +++ b/apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/tests/LsuIntegrationTest.scala @@ -297,15 +297,9 @@ class LsuIntegrationTest inside(sv1ScanBackend.listDsoSequencers()) { case Seq(DomainSequencers(synchronizerId, sequencers)) => synchronizerId shouldBe decentralizedSynchronizerId - sequencers should have size 8 + sequencers should have size 4 sequencers.foreach { sequencer => - sequencer.serial match { - case Some(serial) => - serial shouldBe 0 - sequencer.migrationId shouldBe -1 - case None => - sequencer.migrationId shouldBe 0 - } + sequencer.serial shouldBe 0 } } } @@ -548,22 +542,16 @@ class LsuIntegrationTest inside(sv1ScanBackend.listDsoSequencers()) { case Seq(DomainSequencers(synchronizerId, sequencers)) => synchronizerId shouldBe decentralizedSynchronizerId - sequencers should have size 11 + sequencers should have size 7 sequencers.groupBy(_.svName).foreach { case (sv, sequencers) => clue(s"check sequencers for $sv") { forExactly(1, sequencers) { sequencer => - sequencer.serial.value shouldBe 0 - sequencer.migrationId shouldBe -1 + sequencer.serial shouldBe 0 } if (sv != sv4Backend.config.onboarding.value.name) forExactly(1, sequencers) { sequencer => - sequencer.serial.value shouldBe newSynchronizerSerial.value.toLong - sequencer.migrationId shouldBe -1 + sequencer.serial shouldBe newSynchronizerSerial.value.toLong } - forExactly(1, sequencers) { sequencer => - sequencer.serial should be(empty) - sequencer.migrationId shouldBe 0 - } } } } @@ -764,11 +752,9 @@ class LsuIntegrationTest .listDsoSequencers() .loneElement .sequencers - .filter(c => - c.svName == sv1LocalBackend.config.onboarding.value.name && c.serial.isDefined - ) + .filter(c => c.svName == sv1LocalBackend.config.onboarding.value.name) .loneElement - .serial shouldBe Some(2), + .serial shouldBe 2, ) } diff --git a/apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/tests/RollForwardLsuDRIntegrationTest.scala b/apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/tests/RollForwardLsuDRIntegrationTest.scala index cd7a5dabce..5c94cb809c 100644 --- a/apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/tests/RollForwardLsuDRIntegrationTest.scala +++ b/apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/tests/RollForwardLsuDRIntegrationTest.scala @@ -393,21 +393,15 @@ class RollForwardLsuDRIntegrationTest inside(sv1ScanLocalBackend.listDsoSequencers()) { case Seq(DomainSequencers(synchronizerId, sequencers)) => synchronizerId shouldBe decentralizedSynchronizerId - sequencers should have size 12 + sequencers should have size 8 sequencers.groupBy(_.svName).foreach { case (sv, sequencers) => clue(s"check sequencers for $sv") { - sequencers.size shouldBe 3 + sequencers.size shouldBe 2 forExactly(1, sequencers) { sequencer => - sequencer.serial.value shouldBe 0 - sequencer.migrationId shouldBe -1 + sequencer.serial shouldBe 0 } forExactly(1, sequencers) { sequencer => - sequencer.serial.value shouldBe newSynchronizerSerial.value.toLong - sequencer.migrationId shouldBe -1 - } - forExactly(1, sequencers) { sequencer => - sequencer.serial should be(empty) - sequencer.migrationId shouldBe 0 + sequencer.serial shouldBe newSynchronizerSerial.value.toLong } } } diff --git a/apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/tests/RollForwardLsuIntegrationTest.scala b/apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/tests/RollForwardLsuIntegrationTest.scala index d72ab247fc..3af4d9b5e1 100644 --- a/apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/tests/RollForwardLsuIntegrationTest.scala +++ b/apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/tests/RollForwardLsuIntegrationTest.scala @@ -182,13 +182,7 @@ class RollForwardLsuIntegrationTest synchronizerId shouldBe decentralizedSynchronizerId sequencers should have size 8 sequencers.foreach { sequencer => - sequencer.serial match { - case Some(serial) => - serial shouldBe 0 - sequencer.migrationId shouldBe -1 - case None => - sequencer.migrationId shouldBe 0 - } + sequencer.serial shouldBe 0 } } } @@ -211,7 +205,7 @@ class RollForwardLsuIntegrationTest clue("Topology state contains LSU announcement") { eventually(3.minutes) { sv1ScanBackend.participantClient.topology.lsu.announcement - .list(Some(decentralizedSynchronizerId)) should have size (1) + .list(Some(decentralizedSynchronizerId)) should have size 1 } } val allSvBackends = Seq(sv1Backend, sv2Backend, sv3Backend, sv4Backend) @@ -349,21 +343,15 @@ class RollForwardLsuIntegrationTest inside(sv1ScanLocalBackend.listDsoSequencers()) { case Seq(DomainSequencers(synchronizerId, sequencers)) => synchronizerId shouldBe decentralizedSynchronizerId - sequencers should have size 12 + sequencers should have size 8 sequencers.groupBy(_.svName).foreach { case (sv, sequencers) => clue(s"check sequencers for $sv") { sequencers.size shouldBe 3 forExactly(1, sequencers) { sequencer => - sequencer.serial.value shouldBe 0 - sequencer.migrationId shouldBe -1 - } - forExactly(1, sequencers) { sequencer => - sequencer.serial.value shouldBe newSynchronizerSerial.value.toLong - sequencer.migrationId shouldBe -1 + sequencer.serial shouldBe 0 } forExactly(1, sequencers) { sequencer => - sequencer.serial should be(empty) - sequencer.migrationId shouldBe 0 + sequencer.serial shouldBe newSynchronizerSerial.value.toLong } } } diff --git a/apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/tests/SvInitializationIntegrationTest.scala b/apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/tests/SvInitializationIntegrationTest.scala index 75b2d420ed..d32268edd5 100644 --- a/apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/tests/SvInitializationIntegrationTest.scala +++ b/apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/tests/SvInitializationIntegrationTest.scala @@ -172,18 +172,6 @@ class SvInitializationIntegrationTest extends SvIntegrationTestBase { "http://localhost:5408", ) } - clue("backwards compatible synchronizers configs is set") { - val allSequencerConfigs = nodeStates.map(_.sequencer.toScala.value) - allSequencerConfigs.size shouldBe 4 - allSequencerConfigs.map(_.migrationId).toSeq.distinct.loneElement shouldBe java.lang.Long - .valueOf(0) - allSequencerConfigs.map(_.url) should contain theSameElementsAs Seq( - "http://localhost:5108", - "http://localhost:5208", - "http://localhost:5308", - "http://localhost:5408", - ) - } } finally { // Remove the sequencer again, otherwise the logic for resetting the namespace to only contain // sv1 will fail. diff --git a/apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/tests/ValidatorSequencerConnectionIntegrationTest.scala b/apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/tests/ValidatorSequencerConnectionIntegrationTest.scala index 8828ea149b..141cbb38f6 100644 --- a/apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/tests/ValidatorSequencerConnectionIntegrationTest.scala +++ b/apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/tests/ValidatorSequencerConnectionIntegrationTest.scala @@ -15,7 +15,6 @@ import scala.concurrent.{ExecutionContext, Future} import org.lfdecentralizedtrust.splice.store.AppStoreWithIngestion.SpliceLedgerConnectionPriority.Low import org.lfdecentralizedtrust.splice.codegen.java.splice.dso.decentralizedsynchronizer.{ PhysicalSynchronizerNodeConfig, - SequencerConfig, SequencerConnectionConfig, SynchronizerNodeConfig, } @@ -84,7 +83,7 @@ class ValidatorSequencerConnectionIntegrationTest val now = env.environment.clock.now val availableSequencers = for { domain <- allSequencers - sequencer <- domain.sequencers.filter(_.serial.isDefined) + sequencer <- domain.sequencers if sequencer.url.nonEmpty && !now.toInstant.isBefore(sequencer.availableAfter) } yield sequencer availableSequencers.size shouldBe 4 @@ -219,24 +218,12 @@ class ValidatorSequencerConnectionIntegrationTest sys.error(s"No config found for synchronizer $synchronizerId"), ) - existingSequencerConfig = synchronizerNodeConfig.sequencer.toScala - .getOrElse( - sys.error(s"No sequencer config found for synchronizer $synchronizerId") - ) - - updatedSequencerConfig = new SequencerConfig( - existingSequencerConfig.migrationId, - existingSequencerConfig.sequencerId, - newUrl, - existingSequencerConfig.availableAfter, - ) - newNodeConfig = new SynchronizerNodeConfig( synchronizerNodeConfig.cometBft, - Some(updatedSequencerConfig).toJava, + None.toJava, synchronizerNodeConfig.mediator, synchronizerNodeConfig.scan, - synchronizerNodeConfig.legacySequencerConfig, + None.toJava, synchronizerNodeConfig.sequencerIdentity, synchronizerNodeConfig.physicalSynchronizers.toScala .map( diff --git a/apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/tests/runbook/RunbookSvSequencerInfoPreflightIntegrationTest.scala b/apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/tests/runbook/RunbookSvSequencerInfoPreflightIntegrationTest.scala index 5224d2f2ff..63d6b258bc 100644 --- a/apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/tests/runbook/RunbookSvSequencerInfoPreflightIntegrationTest.scala +++ b/apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/tests/runbook/RunbookSvSequencerInfoPreflightIntegrationTest.scala @@ -26,10 +26,10 @@ class RunbookSvSequencerInfoPreflightIntegrationTest extends IntegrationTest { } val nodeState: SvNodeState = dsoInfo.svNodeStates.get(dsoInfo.svParty).value.payload val domainConfig = nodeState.state.synchronizerNodes.asScala.values.headOption.value - val sequencerUrl = domainConfig.physicalSynchronizers.toScala - .flatMap(_.asScala.get(migrationId).flatMap(_.sequencer.toScala.map(_.url))) - .orElse(domainConfig.sequencer.toScala.filter(_.migrationId == migrationId).map(_.url)) - .value - sequencerUrl shouldBe s"https://sequencer-$migrationId.sv.${sys.env("NETWORK_APPS_ADDRESS")}" + val sequencerUrls = domainConfig.physicalSynchronizers.toScala.toList + .flatMap(_.asScala.values.flatMap(_.sequencer.toScala.map(_.url))) + sequencerUrls should contain( + s"https://sequencer-$migrationId.sv.${sys.env("NETWORK_APPS_ADDRESS")}" + ) } } diff --git a/apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/tests/runbook/ValidatorPreflightIntegrationTest.scala b/apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/tests/runbook/ValidatorPreflightIntegrationTest.scala index a07d90536c..187cabaa55 100644 --- a/apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/tests/runbook/ValidatorPreflightIntegrationTest.scala +++ b/apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/tests/runbook/ValidatorPreflightIntegrationTest.scala @@ -464,19 +464,12 @@ abstract class ValidatorPreflightIntegrationTestBase env.environment.clock.now.toInstant.isAfter(connection.availableAfter.plusSeconds(60)) } - val availableConnections = if (connections.forall(_.serial.isEmpty)) { - val latestMigrationId = connections.map(_.migrationId).max - connections.filter(connection => - connection.migrationId == latestMigrationId && - connection.url != "" && - isAvailable(connection) - ) - } else { - connections.filter(connection => - connection.serial.contains(migrationId) && - isAvailable(connection) - ) - } + val latestSerial = connections.map(_.serial).max + val availableConnections = connections.filter(connection => + connection.serial == latestSerial && + connection.url != "" && + isAvailable(connection) + ) val (expectedSequencerConnections, _) = Endpoint .fromUris(NonEmpty.from(availableConnections.map(conn => new URI(conn.url))).value) @@ -633,10 +626,12 @@ class RunbookValidatorPreflightIntegrationTest extends ValidatorPreflightIntegra val synchronizerNodeConfig = nodeState.state.synchronizerNodes.asScala.values.headOption.value val svSequencerUrl = synchronizerNodeConfig.physicalSynchronizers.toScala - .flatMap(_.asScala.get(migrationId).flatMap(_.sequencer.map(_.url).toScala)) - .getOrElse( - synchronizerNodeConfig.sequencer.toScala.value.url + .flatMap( + _.asScala.toSeq + .maxByOption(_._1.longValue()) + .flatMap(_._2.sequencer.map(_.url).toScala) ) + .value val (svSequencerEndpoint, _) = Endpoint .fromUris( NonEmpty.from(Seq(new URI(svSequencerUrl))).value diff --git a/apps/common/src/main/scala/org/lfdecentralizedtrust/splice/environment/PackageVersionSupport.scala b/apps/common/src/main/scala/org/lfdecentralizedtrust/splice/environment/PackageVersionSupport.scala index d349c5207a..b42062a9df 100644 --- a/apps/common/src/main/scala/org/lfdecentralizedtrust/splice/environment/PackageVersionSupport.scala +++ b/apps/common/src/main/scala/org/lfdecentralizedtrust/splice/environment/PackageVersionSupport.scala @@ -130,19 +130,6 @@ trait PackageVersionSupport extends NamedLogging { ) } - // TODO(#564) - ensure the right version is used - def supportsPhysicalSynchronizers(parties: Seq[PartyId], now: CantonTimestamp)(implicit - tc: TraceContext - ): Future[FeatureSupport] = { - isDarSupported( - parties, - PackageIdResolver.Package.SpliceDsoGovernance, - now, - DarResources.dsoGovernance, - DarResources.dsoGovernance_0_1_24, - ) - } - def supportsTrafficBasedAppRewards(parties: Seq[PartyId], now: CantonTimestamp)(implicit tc: TraceContext ): Future[FeatureSupport] = diff --git a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/admin/api/client/commands/HttpScanAppClient.scala b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/admin/api/client/commands/HttpScanAppClient.scala index b794cdd540..b823c77ca6 100644 --- a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/admin/api/client/commands/HttpScanAppClient.scala +++ b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/admin/api/client/commands/HttpScanAppClient.scala @@ -784,15 +784,16 @@ object HttpScanAppClient { Codec.decode(Codec.SynchronizerId)(domain.domainId).flatMap { synchronizerId => domain.sequencers .traverse { s => - Codec.decode(Codec.Sequencer)(s.id).map { sequencerId => - DsoSequencer( - s.migrationId, - s.synchronizerSerial, - sequencerId, - s.url, - s.svName, - s.availableAfter.toInstant, - ) + Codec.decode(Codec.Sequencer)(s.id).flatMap { sequencerId => + s.synchronizerSerial.toRight("No serial provided").map { serial => + DsoSequencer( + serial, + sequencerId, + s.url, + s.svName, + s.availableAfter.toInstant, + ) + } } } .map { sequencers => @@ -806,8 +807,7 @@ object HttpScanAppClient { final case class DomainSequencers(synchronizerId: SynchronizerId, sequencers: Seq[DsoSequencer]) final case class DsoSequencer( - migrationId: Long, - serial: Option[Long], + serial: Long, id: SequencerId, url: String, svName: String, diff --git a/apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/lsu/LsuTrigger.scala b/apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/lsu/LsuTrigger.scala index 966bd3b50a..464b0b93ce 100644 --- a/apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/lsu/LsuTrigger.scala +++ b/apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/lsu/LsuTrigger.scala @@ -3,7 +3,7 @@ package org.lfdecentralizedtrust.splice.sv.lsu -import cats.implicits.{catsSyntaxOptionId, showInterpolator, toTraverseOps} +import cats.implicits.{showInterpolator, toTraverseOps} import com.digitalasset.canton.admin.api.client.data.NodeStatus import com.digitalasset.canton.data.CantonTimestamp import com.digitalasset.canton.logging.pretty.{Pretty, PrettyPrinting} @@ -170,7 +170,7 @@ class LsuTrigger( Future.unit } _ <- reconciler.reconcileSynchronizerNodeConfigIfRequired( - localSynchronizerNodes.some, + localSynchronizerNodes, currentPsid.logical, OnboardedImmediately, ) diff --git a/apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/onboarding/SynchronizerNodeReconciler.scala b/apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/onboarding/SynchronizerNodeReconciler.scala index 5919e636c3..a235a9a6f9 100644 --- a/apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/onboarding/SynchronizerNodeReconciler.scala +++ b/apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/onboarding/SynchronizerNodeReconciler.scala @@ -12,21 +12,14 @@ import com.digitalasset.canton.tracing.TraceContext import com.digitalasset.canton.util.MonadUtil import com.digitalasset.canton.util.ShowUtil.* import org.lfdecentralizedtrust.splice.codegen.java.splice.dso.decentralizedsynchronizer.{ - LegacySequencerConfig, MediatorConfig, PhysicalSynchronizerNodeConfig, ScanConfig, - SequencerConfig, SequencerConnectionConfig, SequencerIdentityConfig, SynchronizerNodeConfig, } -import org.lfdecentralizedtrust.splice.environment.{ - PackageVersionSupport, - RetryFor, - RetryProvider, - SpliceLedgerConnection, -} +import org.lfdecentralizedtrust.splice.environment.{RetryFor, RetryProvider, SpliceLedgerConnection} import org.lfdecentralizedtrust.splice.environment.SynchronizerNode.LocalSynchronizerNodes import org.lfdecentralizedtrust.splice.store.DsoRulesStore.DsoRulesWithSvNodeState import org.lfdecentralizedtrust.splice.sv.config.SvScanConfig @@ -34,7 +27,7 @@ import org.lfdecentralizedtrust.splice.sv.onboarding.SynchronizerNodeReconciler. import org.lfdecentralizedtrust.splice.sv.store.SvDsoStore import org.lfdecentralizedtrust.splice.sv.LocalSynchronizerNode import org.lfdecentralizedtrust.splice.sv.util.SvUtil -import org.lfdecentralizedtrust.splice.sv.util.SvUtil.{LocalMediatorConfig, LocalSequencerConfig} +import org.lfdecentralizedtrust.splice.sv.util.SvUtil.LocalMediatorConfig import org.lfdecentralizedtrust.splice.util.PrettyInstances.* import java.lang @@ -46,11 +39,9 @@ import scala.jdk.OptionConverters.{RichOption, RichOptional} class SynchronizerNodeReconciler( dsoStore: SvDsoStore, connection: SpliceLedgerConnection, - versionSupport: PackageVersionSupport, clock: Clock, retryProvider: RetryProvider, val loggerFactory: NamedLoggerFactory, - migrationId: Long, scanConfig: SvScanConfig, ) extends NamedLogging { @@ -58,52 +49,35 @@ class SynchronizerNodeReconciler( private val dsoParty = dsoStore.key.dsoParty def reconcileSynchronizerNodeConfigIfRequired( - synchronizerNodes: Option[LocalSynchronizerNodes[LocalSynchronizerNode]], + synchronizerNodes: LocalSynchronizerNodes[LocalSynchronizerNode], synchronizerId: SynchronizerId, state: SynchronizerNodeState, )(implicit ec: ExecutionContext, tc: TraceContext, ): Future[Unit] = { - val currentNode = synchronizerNodes.map(_.current) + val currentNode = synchronizerNodes.current def setConfigIfRequired() = for { - localSequencerConfig <- SvUtil.getSequencerConfig(currentNode, migrationId) - localMediatorConfig <- SvUtil.getMediatorConfig(currentNode) + localSequencerConfig <- SvUtil.getSequencerConfig(Some(currentNode)) + localMediatorConfig <- SvUtil.getMediatorConfig(Some(currentNode)) localScanConfig = java.util.Optional.of(new ScanConfig(scanConfig.publicUrl.toString())) rulesAndState <- dsoStore.getDsoRulesWithSvNodeState(svParty) nodeState = rulesAndState.svNodeState.payload - // TODO(DACH-NY/canton-network-node#4901): do not use default, but reconcile all configured domains synchronizerNodeConfig = nodeState.state.synchronizerNodes.asScala .get(synchronizerId.toProtoPrimitive) - sequencerConfig = synchronizerNodeConfig.flatMap(_.sequencer.toScala) existingSequencerIdentityConfig = synchronizerNodeConfig.flatMap(_.sequencerIdentity.toScala) mediatorConfig = synchronizerNodeConfig.flatMap(_.mediator.toScala) existingScanConfig = synchronizerNodeConfig.flatMap(_.scan.toScala).toJava - existingSequencerConfig = sequencerConfig.map(c => - LocalSequencerConfig(c.sequencerId, c.url, c.migrationId) - ) existingMediatorConfig = mediatorConfig.map(c => LocalMediatorConfig(c.mediatorId)) - existingLegacySequencerConfig = synchronizerNodeConfig.flatMap( - _.legacySequencerConfig.toScala - ) shouldMarkSequencerAsOnboarded = state match { case SynchronizerNodeState.OnboardedAfterDelay | SynchronizerNodeState.OnboardedImmediately => existingSequencerIdentityConfig .flatMap(_.availableAfter.toScala) - .orElse(sequencerConfig.flatMap(_.availableAfter.toScala)) .isEmpty case SynchronizerNodeState.Onboarding(_) => false } - updatedSequencerConfigUpdate = - updateLegacySequencerConfig( - existingLegacySequencerConfig - ) - _ = ensureSequencerUrlIsDifferentWhenSynchronizerUpgraded( - existingSequencerConfig, - localSequencerConfig, - ) existingPhysicalSynchronizers = synchronizerNodeConfig.flatMap( _.physicalSynchronizers.toScala ) @@ -113,13 +87,17 @@ class SynchronizerNodeReconciler( scalaExistingPhysicalSynchronizers, state, ) + sequencerIdentityChanged = existingSequencerIdentityConfig.map( + _.sequencerId + ) != localSequencerConfig.map( + _.sequencerId + ) _ <- if ( - existingSequencerConfig != localSequencerConfig || + sequencerIdentityChanged || existingMediatorConfig != localMediatorConfig || existingScanConfig != localScanConfig || shouldMarkSequencerAsOnboarded || - updatedSequencerConfigUpdate.isRight || scalaExistingPhysicalSynchronizers != localPhysicalSynchronizers ) { def setConfig( @@ -144,13 +122,7 @@ class SynchronizerNodeReconciler( val sequencerAvailableAfter: Option[Instant] = localSequencerConfig.flatMap { _ => val sequencerAvailabilityDelay = - currentNode - .map(_.sequencerAvailabilityDelay) - .getOrElse( - sys.error( - "synchronizerNode is not expected to be empty." - ) - ) + currentNode.sequencerAvailabilityDelay state match { case SynchronizerNodeState.OnboardedAfterDelay => Some(clock.now.toInstant.plus(sequencerAvailabilityDelay)) @@ -161,38 +133,23 @@ class SynchronizerNodeReconciler( } } - // When physical synchronizers with sequencerIdentity is supported, - // prefer setting sequencerIdentity over the legacy sequencer field. val sequencerIdentityConfig: java.util.Optional[SequencerIdentityConfig] = - if (localPhysicalSynchronizers.isDefined) { - localSequencerConfig - .map(c => - new SequencerIdentityConfig( - c.sequencerId, - existingSequencerIdentityConfig - .flatMap(_.availableAfter.toScala) - .orElse(sequencerAvailableAfter) - .toJava, - ) + localSequencerConfig + .map(c => + new SequencerIdentityConfig( + c.sequencerId, + existingSequencerIdentityConfig + .flatMap(_.availableAfter.toScala) + .orElse(sequencerAvailableAfter) + .toJava, ) - .toJava - } else { - synchronizerNodeConfig.flatMap(_.sequencerIdentity.toScala).toJava - } + ) + .toJava val nodeConfig = new SynchronizerNodeConfig( synchronizerNodeConfig.map(_.cometBft).getOrElse(SvUtil.emptyCometBftConfig), - localSequencerConfig.map { c => - new SequencerConfig( - c.migrationId, - c.sequencerId, - c.url, - sequencerConfig - .flatMap(_.availableAfter.toScala) - .orElse(sequencerAvailableAfter) - .toJava, - ) - }.toJava, + // deprecated in favor of the sequencerIdentityConfig and physicalSynchronizers + None.toJava, localMediatorConfig .map(c => new MediatorConfig( @@ -201,13 +158,14 @@ class SynchronizerNodeReconciler( ) .toJava, localScanConfig, - updatedSequencerConfigUpdate.getOrElse(existingLegacySequencerConfig).toJava, + // deprecated legacy sequencer config + None.toJava, sequencerIdentityConfig, localPhysicalSynchronizers .map(_.asJava) .toJava, ) - setConfig(synchronizerId, rulesAndState, nodeConfig) + setConfig(synchronizerId.logical, rulesAndState, nodeConfig) } else { logger.info(s"Not setting domain node config because it is the same as the existing one.") Future.unit @@ -225,77 +183,81 @@ class SynchronizerNodeReconciler( } private[onboarding] def buildPhysicalSynchronizers( - synchronizerNodes: Option[LocalSynchronizerNodes[LocalSynchronizerNode]], + synchronizerNodes: LocalSynchronizerNodes[LocalSynchronizerNode], existingState: Option[Map[lang.Long, PhysicalSynchronizerNodeConfig]], state: SynchronizerNodeState, )(implicit ec: ExecutionContext, tc: TraceContext, ): Future[Option[Map[lang.Long, PhysicalSynchronizerNodeConfig]]] = { - versionSupport - .supportsPhysicalSynchronizers(Seq(svParty, dsoParty), clock.now) - .map(_.supported) - .flatMap { hasSupport => - if (hasSupport) { - val currentEntryFuture = - synchronizerNodes.map(_.current).traverse { currentNode => - val serialOverride = state match { - case SynchronizerNodeState.OnboardedAfterDelay => None - case SynchronizerNodeState.OnboardedImmediately => None - case SynchronizerNodeState.Onboarding(serial) => Some(serial) - } - buildNodeConfig(currentNode, serialOverride) - } + val currentEntryFuture = { + val currentNode = synchronizerNodes.current + val serialOverride = state match { + case SynchronizerNodeState.OnboardedAfterDelay => None + case SynchronizerNodeState.OnboardedImmediately => None + case SynchronizerNodeState.Onboarding(serial) => Some(serial) + } + buildNodeConfig(currentNode, serialOverride) + } - val legacyEntryFuture = - synchronizerNodes.flatMap(_.legacy).traverse { legacyNode => - buildNodeConfig(legacyNode) - } + val legacyEntryFuture = + synchronizerNodes.legacy.traverse { legacyNode => + buildNodeConfig(legacyNode) + } - val additionalLegacyEntriesFuture = - MonadUtil.sequentialTraverse(synchronizerNodes.toList.flatMap(_.additionalLegacy)) { - legacyNode => - buildNodeConfig(legacyNode) - } + val additionalLegacyEntriesFuture = + MonadUtil.sequentialTraverse(synchronizerNodes.additionalLegacy.toList) { legacyNode => + buildNodeConfig(legacyNode) + } - val successorEntryFuture = - synchronizerNodes.flatMap(_.successor).flatTraverse { successorNode => - successorNode.sequencerAdminConnection - .isNodeInitialized() - .attemptT - .foldF( - failure => - currentEntryFuture.map(_.flatMap { case (currentSyncSerial, _) => - val existingSuccessors = - existingState.map(_.view.filterKeys(_ > currentSyncSerial).toSeq) - logger.info( - s"Failed to get successor status, will keep state with serial > than $currentSyncSerial: $existingSuccessors", - failure, - ) - existingSuccessors - }), - { - case true => - buildNodeConfig(successorNode).map(config => Some(Seq(config))) - case false => - Future.successful(None) - }, + val successorEntryFuture = + synchronizerNodes.successor.flatTraverse { successorNode => + successorNode.sequencerAdminConnection + .isNodeInitialized() + .attemptT + .foldF( + failure => + currentEntryFuture.map { case (currentSyncSerial, _) => + val existingSuccessors = + existingState.map(_.view.filterKeys(_ > currentSyncSerial).toSeq) + logger.info( + s"Failed to get successor status, will keep state with serial > than $currentSyncSerial: $existingSuccessors", + failure, ) - } - - for { - currentEntry <- currentEntryFuture - legacyEntry <- legacyEntryFuture - successorEntry <- successorEntryFuture - additionalLegacyEntries <- additionalLegacyEntriesFuture - } yield { - Some( - (legacyEntry.toList ++ currentEntry.toList ++ successorEntry.toList.flatten ++ additionalLegacyEntries).toMap - ) - } - } else Future.successful(None) + existingSuccessors + }, + { + case true => + buildNodeConfig(successorNode).map(config => Some(Seq(config))) + case false => + Future.successful(None) + }, + ) } + for { + currentEntry <- currentEntryFuture + legacyEntry <- legacyEntryFuture + successorEntry <- successorEntryFuture + additionalLegacyEntries <- additionalLegacyEntriesFuture + } yield { + val entries = + legacyEntry.toList ++ List( + currentEntry + ) ++ successorEntry.toList.flatten ++ additionalLegacyEntries + val urlToSerials = entries + .groupMap { case (_, config) => + config.sequencer.toScala.map(_.url) + } { case (serial, _) => serial } + urlToSerials.foreach { case (url, serials) => + if (serials.distinct.sizeIs > 1) { + sys.error( + s"Different serials ${serials.distinct} are configured with the same sequencer url $url" + ) + } + } + Some(entries.toMap) + } } private def buildNodeConfig( @@ -323,33 +285,6 @@ class SynchronizerNodeReconciler( } } - private def ensureSequencerUrlIsDifferentWhenSynchronizerUpgraded( - existingSequencerConfigOpt: Option[LocalSequencerConfig], - sequencerConfigOpt: Option[LocalSequencerConfig], - ): Unit = { - if ( - existingSequencerConfigOpt.exists { existingSequencerConfig => - sequencerConfigOpt.exists(sequencerConfig => - existingSequencerConfig.migrationId != sequencerConfig.migrationId && - existingSequencerConfig.url == sequencerConfig.url - ) - } - ) - sys.error("Sequencer URL must be different when domain is upgraded.") - } - private def updateLegacySequencerConfig( - existingLegacySequencerConfig: Option[LegacySequencerConfig] - )(implicit - tc: TraceContext - ): Either[Unit, Option[LegacySequencerConfig]] = - if (existingLegacySequencerConfig.isDefined) { - logger.info( - s"removing existing legacy sequencer config as there is no expected legacy migration id" - ) - Right(None) - } else { - Left(()) - } } object SynchronizerNodeReconciler { diff --git a/apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/onboarding/joining/JoiningNodeInitializer.scala b/apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/onboarding/joining/JoiningNodeInitializer.scala index af9fe041ad..d7ea60056e 100644 --- a/apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/onboarding/joining/JoiningNodeInitializer.scala +++ b/apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/onboarding/joining/JoiningNodeInitializer.scala @@ -277,11 +277,9 @@ class JoiningNodeInitializer( synchronizerNodeReconciler = new SynchronizerNodeReconciler( dsoStore, connection, - packageVersionSupport, clock, retryProvider, loggerFactory, - domainMigrationId, config.scan, ) dsoAutomation = @@ -490,7 +488,7 @@ class JoiningNodeInitializer( // This triggers automation in other SV apps, that's why we make sure the sequencer is known first preInit = () => synchronizerNodeReconciler.reconcileSynchronizerNodeConfigIfRequired( - Some(synchronizerNodeService.nodes), + synchronizerNodeService.nodes, decentralizedSynchronizer, Onboarding(participantReportedPSid.serial), ), @@ -522,7 +520,7 @@ class JoiningNodeInitializer( if (!config.shouldSkipSynchronizerInitialization) { synchronizerNodeReconciler .reconcileSynchronizerNodeConfigIfRequired( - synchronizerNodeService.nodes.some, + synchronizerNodeService.nodes, decentralizedSynchronizer, OnboardedAfterDelay, ) @@ -884,15 +882,12 @@ class JoiningNodeInitializer( svStore.key.dsoParty, ) _ = logger.info(s"granted ${config.ledgerApiUser} readAs rights for dsoParty") - domainMigrationId <- resolveDomainMigrationId(migrationIdFromSponsorSv()) synchronizerNodeReconciler = new SynchronizerNodeReconciler( dsoStore, svStoreWithIngestion.connection(SpliceLedgerConnectionPriority.Low), - packageVersionSupport, clock, retryProvider, loggerFactory, - domainMigrationId, config.scan, ) dsoAutomation = newSvDsoAutomationService( diff --git a/apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/onboarding/sv1/SV1Initializer.scala b/apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/onboarding/sv1/SV1Initializer.scala index 17e1311489..6cbfb0221d 100644 --- a/apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/onboarding/sv1/SV1Initializer.scala +++ b/apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/onboarding/sv1/SV1Initializer.scala @@ -4,7 +4,6 @@ package org.lfdecentralizedtrust.splice.sv.onboarding.sv1 import cats.implicits.{ - catsSyntaxOptionId, catsSyntaxTuple2Semigroupal, catsSyntaxTuple3Semigroupal, catsSyntaxTuple4Semigroupal, @@ -259,7 +258,8 @@ class SV1Initializer( participantId, dsoAcsStoreDescriptorUserVersion, ) - synchronizerId <- participantAdminConnection.getSynchronizerId(config.domains.global.alias) + psid <- participantAdminConnection.getPhysicalSynchronizerId(config.domains.global.alias) + synchronizerId = psid.logical packageVersionSupport = PackageVersionSupport.createPackageVersionSupport( synchronizerId, initConnection, @@ -322,11 +322,9 @@ class SV1Initializer( new SynchronizerNodeReconciler( dsoStore, connection, - packageVersionSupport, clock, retryProvider, loggerFactory, - domainMigrationId, config.scan, ), ) @@ -334,9 +332,7 @@ class SV1Initializer( _ <- dsoStore.domains.waitForDomainConnection(config.domains.global.alias) withDsoStore = new WithDsoStore( dsoAutomation, - decentralizedSynchronizer, - packageVersionSupport, - domainMigrationId, + psid, ) _ <- retryProvider.ensureThatB( RetryFor.WaitingOnInitDependency, @@ -589,11 +585,10 @@ class SV1Initializer( */ private class WithDsoStore( dsoStoreWithIngestion: AppStoreWithIngestion[SvDsoStore], - synchronizerId: SynchronizerId, - packageVersionSupport: PackageVersionSupport, - domainMigrationId: Long, + psid: PhysicalSynchronizerId, ) { + private val synchronizerId = psid.logical private val dsoStore = dsoStoreWithIngestion.store private val dsoParty = dsoStore.key.dsoParty private val svParty = dsoStore.key.svParty @@ -602,8 +597,6 @@ class SV1Initializer( dsoStoreWithIngestion.connection(SpliceLedgerConnectionPriority.Low), clock = clock, retryProvider = retryProvider, - versionSupport = packageVersionSupport, - migrationId = domainMigrationId, scanConfig = config.scan, loggerFactory = loggerFactory, ) @@ -625,7 +618,7 @@ class SV1Initializer( tc: TraceContext ): Future[Unit] = { synchronizerNodeReconciler.reconcileSynchronizerNodeConfigIfRequired( - synchronizerNodeService.nodes.some, + synchronizerNodeService.nodes, synchronizerId, SynchronizerNodeState.OnboardedImmediately, ) @@ -685,9 +678,8 @@ class SV1Initializer( synchronizerNodeService.nodes.current.cometbftNode, synchronizerNodeService.nodes.current, config.scan, - synchronizerId, + psid, clock, - domainMigrationId, ) _ = logger .info( diff --git a/apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/util/SvUtil.scala b/apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/util/SvUtil.scala index 42ed608cf6..2231f0f030 100644 --- a/apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/util/SvUtil.scala +++ b/apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/util/SvUtil.scala @@ -12,8 +12,10 @@ import org.lfdecentralizedtrust.splice.codegen.java.splice.cometbft.{ import org.lfdecentralizedtrust.splice.codegen.java.splice.dso.decentralizedsynchronizer.{ DsoDecentralizedSynchronizerConfig, MediatorConfig, + PhysicalSynchronizerNodeConfig, ScanConfig, - SequencerConfig, + SequencerConnectionConfig, + SequencerIdentityConfig, SynchronizerConfig, SynchronizerNodeConfig, SynchronizerNodeConfigLimits, @@ -29,7 +31,7 @@ import com.digitalasset.canton.config.{NonNegativeFiniteDuration, PositiveDurati import com.digitalasset.canton.logging.TracedLogger import com.digitalasset.canton.protocol.AcsCommitmentsCatchUpParameters import com.digitalasset.canton.time.Clock -import com.digitalasset.canton.topology.{PartyId, SynchronizerId} +import com.digitalasset.canton.topology.{PartyId, PhysicalSynchronizerId, SynchronizerId} import com.digitalasset.canton.tracing.TraceContext import java.security.interfaces.{ECPrivateKey, ECPublicKey} @@ -141,10 +143,9 @@ object SvUtil { case class LocalSequencerConfig( sequencerId: String, url: String, - migrationId: Long, ) - def getSequencerConfig(synchronizerNode: Option[SvSynchronizerNode], migrationId: Long)(implicit + def getSequencerConfig(synchronizerNode: Option[SvSynchronizerNode])(implicit ec: ExecutionContext, tc: TraceContext, ): Future[Option[LocalSequencerConfig]] = synchronizerNode.map { node => @@ -152,7 +153,6 @@ object SvUtil { LocalSequencerConfig( sequencerId.toProtoPrimitive, node.sequencerExternalPublicUrl, - migrationId, ) } }.sequence @@ -174,9 +174,8 @@ object SvUtil { cometBftNode: Option[CometBftNode], localSynchronizerNode: LocalSynchronizerNode, scanConfig: SvScanConfig, - synchronizerId: SynchronizerId, + synchronizerId: PhysicalSynchronizerId, clock: Clock, - migrationId: Long, )(implicit ec: ExecutionContext, tc: TraceContext, @@ -206,12 +205,10 @@ object SvUtil { ) } .getOrElse(SvUtil.emptyCometBftConfig) - localSequencerConfig <- getSequencerConfig(Some(localSynchronizerNode), migrationId) + localSequencerConfig <- getSequencerConfig(Some(localSynchronizerNode)) sequencerConfig = localSequencerConfig.map(c => - new SequencerConfig( - migrationId, + new SequencerIdentityConfig( c.sequencerId, - c.url, Some(clock.now.toInstant).toJava, ) ) @@ -223,14 +220,21 @@ object SvUtil { ) } yield { Map( - synchronizerId.toProtoPrimitive -> new SynchronizerNodeConfig( + synchronizerId.logical.toProtoPrimitive -> new SynchronizerNodeConfig( cometBftConfig, - sequencerConfig.toJava, + Optional.empty(), mediatorConfig.toJava, Optional.of(new ScanConfig(scanConfig.publicUrl.toString())), Optional.empty(), - Optional.empty(), - Optional.empty(), + sequencerConfig.toJava, + Optional.of( + Map( + java.lang.Long.valueOf(synchronizerId.serial.value.toLong) -> + new PhysicalSynchronizerNodeConfig( + localSequencerConfig.map(c => new SequencerConnectionConfig(c.url)).toJava + ) + ).asJava + ), ) ).asJava } diff --git a/apps/validator/src/main/scala/org/lfdecentralizedtrust/splice/validator/ValidatorApp.scala b/apps/validator/src/main/scala/org/lfdecentralizedtrust/splice/validator/ValidatorApp.scala index 1fe16ae495..047392bf0b 100644 --- a/apps/validator/src/main/scala/org/lfdecentralizedtrust/splice/validator/ValidatorApp.scala +++ b/apps/validator/src/main/scala/org/lfdecentralizedtrust/splice/validator/ValidatorApp.scala @@ -192,7 +192,6 @@ class ValidatorApp( config, participantAdminConnection, scanConnection, - domainMigrationId, retryProvider, loggerFactory, ) @@ -765,7 +764,6 @@ class ValidatorApp( config, participantAdminConnection, scanConnection, - domainMigrationId, retryProvider, loggerFactory, ), diff --git a/apps/validator/src/main/scala/org/lfdecentralizedtrust/splice/validator/domain/DomainConnector.scala b/apps/validator/src/main/scala/org/lfdecentralizedtrust/splice/validator/domain/DomainConnector.scala index 6e19816ddc..c444367acf 100644 --- a/apps/validator/src/main/scala/org/lfdecentralizedtrust/splice/validator/domain/DomainConnector.scala +++ b/apps/validator/src/main/scala/org/lfdecentralizedtrust/splice/validator/domain/DomainConnector.scala @@ -36,7 +36,6 @@ class DomainConnector( config: ValidatorAppBackendConfig, participantAdminConnection: ParticipantAdminConnection, scanConnection: BftScanConnection, - migrationId: Long, retryProvider: RetryProvider, val loggerFactory: NamedLoggerFactory, )(implicit ec: ExecutionContext) @@ -81,7 +80,7 @@ class DomainConnector( ) } - def getDecentralizedSynchronizerSequencerConnections(clock: Clock)(implicit + private def getDecentralizedSynchronizerSequencerConnections(clock: Clock)(implicit tc: TraceContext ): Future[Map[SynchronizerAlias, SequencerConnections]] = { config.domains.global.url match { @@ -191,7 +190,7 @@ class DomainConnector( if (connections.isEmpty) { throw Status.NOT_FOUND .withDescription( - s"sequencer connections for migration id $migrationId and serial $synchronizerSerial is empty at $time, validate with your SV sponsor that your migration id is correct" + s"sequencer connections for serial $synchronizerSerial is empty at $time" ) .asRuntimeException() } else { @@ -200,7 +199,7 @@ class DomainConnector( case None => throw Status.NOT_FOUND .withDescription( - s"sequencer connections for migration id $migrationId and serial $synchronizerSerial is empty at $time, validate with your SV sponsor that your migration id is correct" + s"sequencer connections for serial $synchronizerSerial is empty at $time" ) .asRuntimeException() case Some(nonEmptyConnections) => @@ -263,35 +262,18 @@ class DomainConnector( sequencers.synchronizerId == decentralizedSynchronizerId ) .map { sequencers => - val serialOrMigrationSequencers = + val serialSequencers = sequencers.sequencers - .groupBy(_.id) - .view - .mapValues { sequencersForId => - val serialMatch = - sequencersForId.find(_.serial.contains(synchronizerSerial.unwrap.toLong)) - // it might be that some SV did not update the url for the latest serial - // in that case we don't want to fallback to the migration id one - // the migration id fallback is valid only if the SV did not sync the per serial urls yet for the first time - val sequencerHasAnyEntryWithSerial = sequencersForId.exists(_.serial.nonEmpty) - if (sequencerHasAnyEntryWithSerial) serialMatch - else - serialMatch.orElse( - sequencersForId.find(s => s.serial.isEmpty && s.migrationId == migrationId) - ) - } - .values - .flatten - .toSeq + .filter(_.serial == synchronizerSerial.unwrap.toLong) val svFilteredSequencers = config.domains.global.trustedSynchronizerConfig match { case Some(config) => val allowedNamesSet = config.svNames.toList.toSet logger.debug( s"Filtering sequencers to only include: ${allowedNamesSet.toList.mkString(", ")}" ) - serialOrMigrationSequencers.filter(s => allowedNamesSet.contains(s.svName)) + serialSequencers.filter(s => allowedNamesSet.contains(s.svName)) case None => - serialOrMigrationSequencers + serialSequencers } val validConnections = extractValidConnections( svFilteredSequencers, @@ -314,7 +296,7 @@ class DomainConnector( // sequencer connections will be ignore if they are with a invalid Alias, empty url or not yet available (`before availableAfter`) sequencers .collect { - case DsoSequencer(_, _, id, url, _, availableAfter) + case DsoSequencer(_, id, url, _, availableAfter) if url.nonEmpty && !domainTime.toInstant .isBefore(availableAfter) => for { diff --git a/apps/validator/src/main/scala/org/lfdecentralizedtrust/splice/validator/lsu/RollForwardLsuTrigger.scala b/apps/validator/src/main/scala/org/lfdecentralizedtrust/splice/validator/lsu/RollForwardLsuTrigger.scala index 9db39af40e..edfeddcb12 100644 --- a/apps/validator/src/main/scala/org/lfdecentralizedtrust/splice/validator/lsu/RollForwardLsuTrigger.scala +++ b/apps/validator/src/main/scala/org/lfdecentralizedtrust/splice/validator/lsu/RollForwardLsuTrigger.scala @@ -96,7 +96,7 @@ final case class RollForwardLsuTrigger( s.sequencers case _ => Seq.empty } - .filter(_.serial == Some(rollForward.successorPhysicalSynchronizerId.serial.unwrap.toLong)) + .filter(_.serial == rollForward.successorPhysicalSynchronizerId.serial.unwrap.toLong) val oldSequencers = config.sequencerConnections.aliasToConnection.forgetNE.values.flatMap(_.sequencerId).toSet val usableNewSequencers = newSequencers.filter(s => oldSequencers.contains(s.id))