diff --git a/apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/tests/NonZeroRoundBootstrapBftTimeBasedIntegrationTest.scala b/apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/tests/NonZeroRoundBootstrapBftTimeBasedIntegrationTest.scala new file mode 100644 index 0000000000..4e97bfd53f --- /dev/null +++ b/apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/tests/NonZeroRoundBootstrapBftTimeBasedIntegrationTest.scala @@ -0,0 +1,278 @@ +package org.lfdecentralizedtrust.splice.integration.tests + +import com.digitalasset.canton.HasExecutionContext +import com.digitalasset.canton.concurrent.Threading +import org.lfdecentralizedtrust.splice.codegen.java.splice.cometbft.{ + CometBftConfig, + CometBftNodeConfig, + GovernanceKeyConfig, + SequencingKeyConfig, +} +import org.lfdecentralizedtrust.splice.codegen.java.splice.dso.decentralizedsynchronizer.{ + ScanConfig, + SynchronizerNodeConfig, +} +import org.lfdecentralizedtrust.splice.codegen.java.splice.dsorules.DsoRules_AddSv +import org.lfdecentralizedtrust.splice.codegen.java.splice.dsorules.actionrequiringconfirmation.ARC_DsoRules +import org.lfdecentralizedtrust.splice.codegen.java.splice.dsorules.dsorules_actionrequiringconfirmation.SRARC_AddSv +import org.lfdecentralizedtrust.splice.codegen.java.splice.types.Round +import org.lfdecentralizedtrust.splice.config.ConfigTransforms +import org.lfdecentralizedtrust.splice.http.v0.definitions +import org.lfdecentralizedtrust.splice.integration.EnvironmentDefinition +import org.lfdecentralizedtrust.splice.integration.tests.SpliceTests.{ + IntegrationTestWithIsolatedEnvironment, + SpliceTestConsoleEnvironment, +} +import org.lfdecentralizedtrust.splice.automation.Trigger +import org.lfdecentralizedtrust.splice.sv.automation.singlesv.onboarding.SvOnboardingUnlimitedTrafficTrigger +import org.lfdecentralizedtrust.splice.util.{TimeTestUtil, WalletTestUtil} + +import java.util.Optional +import scala.jdk.CollectionConverters.* + +/** Tests that SV automation triggers can complete the reward + * pipeline for the initial round when bootstrapping TBAR at a + * non-zero round with BFT f >= 1. + * + * Adds a dummy 5th SV to increase the BFT quorum from 1 to 2. + */ +class NonZeroRoundBootstrapBftTimeBasedIntegrationTest + extends IntegrationTestWithIsolatedEnvironment + with HasExecutionContext + with WalletTestUtil + with TimeTestUtil { + + private val initialRound = 4815L + + override def environmentDefinition: SpliceEnvironmentDefinition = + EnvironmentDefinition + .simpleTopology4SvsWithSimTime(this.getClass.getSimpleName) + .withoutAutomaticRewardsCollectionAndAmuletMerging + .addConfigTransforms((_, config) => + ConfigTransforms.updateAllSvAppFoundDsoConfigs_( + _.copy(initialRound = initialRound) + )(config) + ) + .addConfigTransform((_, config) => ConfigTransforms.withNoVoteCooldown(config)) + + "SV triggers complete the reward pipeline for the initial round" in { implicit env => + import definitions.GetRewardAccountingActivityTotalsResponse.members.RewardAccountingActivityTotalsOk + import definitions.GetRewardAccountingActivityTotalsResponse.members.RewardAccountingActivityTotalsCannotProvide + + // Add a dummy 5th SV with a fake scan URL BEFORE any round + // advancement. This raises BFT f from 0 to 1, so the initial + // round's reward pipeline runs under the higher quorum. + addDummySvWithFakeScanUrl() + + // The dummy SV's participant doesn't exist on the sequencer, so + // this trigger would poll indefinitely logging "does not yet have + // a traffic state". Pause it to avoid log noise on CI. + env.svs.local.foreach( + _.dsoAutomation.trigger[SvOnboardingUnlimitedTrafficTrigger].pause().futureValue + ) + + // With f=1, BFT reads need 2 agreeing Ok responses — but only + // SV1's scan has the initial round's data. The randomSingleCall + // override for the initial round lets each SV read from a single + // peer, so the pipeline completes despite the higher quorum. + advanceTimeForRewardAutomationToRunForCurrentRound + actAndCheck( + "Advance to next round opening", + advanceRoundsToNextRoundOpening, + )( + "SV1's scan has reward totals and the initial round is issuing", + _ => { + sv1ScanBackend + .getRewardAccountingActivityTotals(initialRound) shouldBe a[ + RewardAccountingActivityTotalsOk + ] + + // The round becoming IssuingMiningRound is DSO-level proof: + // under BFT f=1, the SummarizingMiningRoundTrigger on each + // SV must obtain reward accounting totals. For the initial + // round, only SV1 has local data; other SVs fall back to + // BFT read (randomSingleCall). If that path failed, fewer + // than f+1=2 SVs could submit summaries and the round would + // not advance. + val (_, issuingRounds) = sv1ScanBackend.getOpenAndIssuingMiningRounds() + issuingRounds.exists( + _.payload.round.number == initialRound + ) shouldBe true + + // SV2's scan does NOT have local reward activity data for + // the initial round — only SV1's scan seeded it. SV2's SV + // trigger obtained the totals via BFT (randomSingleCall), + // but the scan HTTP endpoint queries the local store, so it + // returns CannotProvide. + sv2ScanBackend + .getRewardAccountingActivityTotals(initialRound) shouldBe a[ + RewardAccountingActivityTotalsCannotProvide + ] + }, + ) + + // Advance two ticks so that verdict ingestion processes batches + // that see OpenMiningRound(initialRound+1) already archived. + // A single tick archives the round, but the verdict batch for + // that tick may be processed before the rewards reference store + // has indexed the archival — so lookupLatestArchivedOpenMiningRound + // returns None and last_archived_round stays at initialRound. + // The second tick generates new verdicts that find the archival + // already indexed, bumping last_archived_round. + advanceTimeAndWaitForRoundOpening + advanceTimeAndWaitForRoundOpening + + // Wait for RewardComputationTrigger to compute totals for the + // next round now that last_archived_round covers it. + eventually() { + sv1ScanBackend + .getRewardAccountingActivityTotals(initialRound + 1) shouldBe a[ + RewardAccountingActivityTotalsOk + ] + } + + // Advance another round to confirm the pipeline continues past + // the initial round under normal BFT (f+1=2) without + // randomSingleCall. + actAndCheck( + "Advance past the initial round", + advanceRoundsToNextRoundOpening, + )( + "SV2's scan has local reward totals for the next round", + _ => + sv2ScanBackend + .getRewardAccountingActivityTotals(initialRound + 1) shouldBe a[ + RewardAccountingActivityTotalsOk + ], + ) + } + + private def addDummySvWithFakeScanUrl()(implicit + env: SpliceTestConsoleEnvironment + ): com.digitalasset.canton.topology.PartyId = { + val dsoInfo = sv1Backend.getDsoInfo() + val svParty = dsoInfo.svParty + val dsoParty = dsoInfo.dsoParty + + // Random suffix avoids collisions with stale parties from + // previous runs (databases persist between local test runs). + val dummySvParty = sv1Backend.participantClientWithAdminToken.ledger_api.parties + .allocate(s"dummy-sv5-${scala.util.Random.nextInt().toHexString}") + .party + + val addSvAction = new ARC_DsoRules( + new SRARC_AddSv( + new DsoRules_AddSv( + dummySvParty.toProtoPrimitive, + "Dummy-SV5", + 1000L, + "PAR::dummy-sv5::dummy", + new Round(initialRound), + ) + ) + ) + + val (_, voteRequest) = actAndCheck( + "sv1 creates vote request to add dummy SV5", + eventuallySucceeds() { + sv1Backend.createVoteRequest( + svParty.toProtoPrimitive, + addSvAction, + "url", + "Add dummy SV5 for BFT threshold test", + sv1Backend.getDsoInfo().dsoRules.payload.config.voteRequestTimeout, + None, + ) + }, + )( + "vote request exists", + _ => sv1Backend.listVoteRequests().loneElement, + ) + + actAndCheck( + "sv2 and sv3 vote yes (3 votes total → executes)", { + Seq(sv2Backend, sv3Backend).foreach { sv => + eventuallySucceeds() { + sv.castVote(voteRequest.contractId, true, "url", "description") + } + } + }, + )( + "dummy SV5 is in DsoRules and raises BFT f from 0 to 1", + _ => { + val info = sv1Backend.getDsoInfo() + info.dsoRules.payload.svs.size() shouldBe 5 + }, + ) + + // Pause ALL delegate-based triggers to prevent DsoRules contract + // churn during the SetSynchronizerNodeConfig submission. + // Sleep briefly after pausing to let in-flight commands complete. + env.svs.local.foreach( + _.dsoDelegateBasedAutomation.triggers[Trigger].foreach(_.pause().futureValue) + ) + Threading.sleep(2000) + actAndCheck( + "Set fake scan URL on dummy SV", + setDummySvScanUrl(dsoParty, dummySvParty), + )( + "Dummy SV's scan URL is in the BFT peer list", + _ => { + val nodeState = sv1Backend.getDsoInfo().svNodeStates(dummySvParty) + nodeState.payload.state.synchronizerNodes.values.asScala + .exists(_.scan.isPresent) shouldBe true + }, + ) + env.svs.local.foreach( + _.dsoDelegateBasedAutomation.triggers[Trigger].foreach(_.resume()) + ) + + dummySvParty + } + + private def setDummySvScanUrl( + dsoParty: com.digitalasset.canton.topology.PartyId, + dummySvParty: com.digitalasset.canton.topology.PartyId, + )(implicit env: SpliceTestConsoleEnvironment): Unit = { + val synchronizerId = decentralizedSynchronizerId.toProtoPrimitive + + val nodeConfig = new SynchronizerNodeConfig( + new CometBftConfig( + Map.empty[String, CometBftNodeConfig].asJava, + Seq.empty[GovernanceKeyConfig].asJava, + Seq.empty[SequencingKeyConfig].asJava, + ), + Optional.empty(), // sequencer + Optional.empty(), // mediator + // Unreachable URL — BFT marks it as a failed peer, increasing + // totalNumber (and thus f) without needing a running scan. + Optional.of(new ScanConfig("http://localhost:1")), + Optional.empty(), // legacySequencerConfig + Optional.empty(), // sequencerIdentity + Optional.empty(), // physicalSynchronizers + ) + + eventuallySucceeds() { + val info = sv1Backend.getDsoInfo() + val currentDsoRulesCid = info.dsoRules.contractId + val currentNodeStateCid = info.svNodeStates + .getOrElse(dummySvParty, fail("SvNodeState not found for dummy SV")) + .contractId + sv1Backend.participantClientWithAdminToken.ledger_api_extensions.commands + .submitJava( + actAs = Seq(dummySvParty), + readAs = Seq(dsoParty), + commands = currentDsoRulesCid + .exerciseDsoRules_SetSynchronizerNodeConfig( + dummySvParty.toProtoPrimitive, + synchronizerId, + nodeConfig, + currentNodeStateCid, + ) + .commands() + .asScala + .toSeq, + ) + } + } +} diff --git a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/admin/api/client/BftScanConnection.scala b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/admin/api/client/BftScanConnection.scala index 3c9c689cda..bd060672b5 100644 --- a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/admin/api/client/BftScanConnection.scala +++ b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/admin/api/client/BftScanConnection.scala @@ -139,6 +139,7 @@ class BftScanConnection( // In the future, before removing disableBackgroundRefresh flag, refactor the bootstrapWithSeedNodes(); // BftCustom should not reuse the AllDsoScansBft class to fetch the initial DsoRules. val disableBackgroundRefresh: Boolean = false, + val initialRound: Option[Long] = None, )(implicit protected val ec: ExecutionContextExecutor, protected val mat: Materializer) extends FlagCloseableAsync with NamedLogging @@ -1015,7 +1016,11 @@ class BftScanConnection( GetRewardAccountingActivityTotalsResponse( RewardAccountingActivityTotalsUndetermined(status = "Undetermined") ) - val callConfig = BftCallConfig.default(scanList.scanConnections) + val callConfig = + if (initialRound.contains(roundNumber)) + BftCallConfig.randomSingleCall(scanList.scanConnections) + else + BftCallConfig.default(scanList.scanConnections) if (!callConfig.enoughAvailableScans) Future.successful(undetermined) else bftCall[RewardAccountingActivityTotalsOk]( @@ -1063,7 +1068,11 @@ class BftScanConnection( GetRewardAccountingRootHashResponse( RewardAccountingRootHashUndetermined(status = "Undetermined") ) - val callConfig = BftCallConfig.default(scanList.scanConnections) + val callConfig = + if (initialRound.contains(roundNumber)) + BftCallConfig.randomSingleCall(scanList.scanConnections) + else + BftCallConfig.default(scanList.scanConnections) if (!callConfig.enoughAvailableScans) Future.successful(undetermined) else bftCall[String]( @@ -2003,6 +2012,7 @@ object BftScanConnection { clock: Clock, retryProvider: RetryProvider, loggerFactory: NamedLoggerFactory, + initialRound: Option[Long] = None, )(implicit ec: ExecutionContextExecutor, tc: TraceContext, @@ -2054,6 +2064,7 @@ object BftScanConnection { clock, retryProvider, loggerFactory, + initialRound = initialRound, ) } diff --git a/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/admin/api/client/BftScanConnectionTest.scala b/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/admin/api/client/BftScanConnectionTest.scala index 889fee4c45..eb50e6873d 100644 --- a/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/admin/api/client/BftScanConnectionTest.scala +++ b/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/admin/api/client/BftScanConnectionTest.scala @@ -225,6 +225,7 @@ class BftScanConnectionTest connectionBuilder: Uri => Future[SingleScanConnection] = _ => Future.failed(new RuntimeException("Shouldn't be refreshing!")), initialFailedConnections: Map[Uri, Throwable] = Map.empty, + initialRound: Option[Long] = None, ) = { new BftScanConnection( mock[SpliceLedgerClient], @@ -242,6 +243,7 @@ class BftScanConnectionTest clock, retryProvider, loggerFactory, + initialRound = initialRound, ) } val notFoundFailure = new BaseAppConnection.UnexpectedHttpJsonResponse( @@ -1346,6 +1348,47 @@ class BftScanConnectionTest ) .map(_ => succeed) } + + "uses randomSingleCall for the initial round" in { + val round = 42L + val connections = getMockedConnections(n = 4) + makeMockReturnRootHashOk(connections(0), round, "aabb") + makeMockReturnRootHashCannotProvide(connections(1), round) + makeMockReturnRootHashCannotProvide(connections(2), round) + makeMockReturnRootHashCannotProvide(connections(3), round) + val bft = getBft(connections, initialRound = Some(round)) + + def attempt(remaining: Int): Future[GetRewardAccountingRootHashResponse] = + bft.getRewardAccountingRootHash(round).flatMap { + case ok: GetRewardAccountingRootHashResponse.members.RewardAccountingRootHashOk => + Future.successful(ok) + case _ if remaining > 1 => attempt(remaining - 1) + case other => Future.successful(other) + } + + for { + resp <- attempt(100) + } yield inside(resp) { + case GetRewardAccountingRootHashResponse.members.RewardAccountingRootHashOk(ok) => + ok.rootHash should be("aabb") + } + } + + "uses default BFT for non-initial rounds even when initialRound is set" in { + val round = 43L + val connections = getMockedConnections(n = 4) + connections.zipWithIndex.foreach { case (c, i) => + makeMockReturnRootHashOk(c, round, s"hash$i") + } + val bft = getBft(connections, initialRound = Some(42L)) + + for { + resp <- bft.getRewardAccountingRootHash(round) + } yield inside(resp) { + case _: GetRewardAccountingRootHashResponse.members.RewardAccountingRootHashUndetermined => + succeed + } + } } "BftScanConnection.getRewardAccountingActivityTotals" should { @@ -1470,6 +1513,48 @@ class BftScanConnectionTest ) .map(_ => succeed) } + + "uses randomSingleCall for the initial round" in { + val round = 42L + val connections = getMockedConnections(n = 4) + makeMockReturnActivityTotalsOk(connections(0), round, 100L, 10L, 5L) + makeMockReturnActivityTotalsCannotProvide(connections(1), round) + makeMockReturnActivityTotalsCannotProvide(connections(2), round) + makeMockReturnActivityTotalsCannotProvide(connections(3), round) + val bft = getBft(connections, initialRound = Some(round)) + + def attempt(remaining: Int): Future[GetRewardAccountingActivityTotalsResponse] = + bft.getRewardAccountingActivityTotals(round).flatMap { + case ok: GetRewardAccountingActivityTotalsResponse.members.RewardAccountingActivityTotalsOk => + Future.successful(ok) + case _ if remaining > 1 => attempt(remaining - 1) + case other => Future.successful(other) + } + + for { + resp <- attempt(100) + } yield inside(resp) { + case GetRewardAccountingActivityTotalsResponse.members + .RewardAccountingActivityTotalsOk(ok) => + ok.totalAppActivityWeight should be(100L) + } + } + + "uses default BFT for non-initial rounds even when initialRound is set" in { + val round = 43L + val connections = getMockedConnections(n = 4) + connections.zipWithIndex.foreach { case (c, i) => + makeMockReturnActivityTotalsOk(c, round, 100L + i, 10L + i, 5L + i) + } + val bft = getBft(connections, initialRound = Some(42L)) + + for { + resp <- bft.getRewardAccountingActivityTotals(round) + } yield inside(resp) { + case _: GetRewardAccountingActivityTotalsResponse.members.RewardAccountingActivityTotalsUndetermined => + succeed + } + } } "BftScanConnection.getRewardAccountingBatch" should { diff --git a/apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/automation/SvDsoAutomationService.scala b/apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/automation/SvDsoAutomationService.scala index b67dafbebb..9a532dff10 100644 --- a/apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/automation/SvDsoAutomationService.scala +++ b/apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/automation/SvDsoAutomationService.scala @@ -188,21 +188,29 @@ class SvDsoAutomationService( case Some(future) => future case None => - val future = BftScanConnection - .peerScanConnection( - () => - BftScanConnection.Bft.getPeerScansFromDsoRules( - dsoStore, - dsoStore.key.svParty, - )(tc, ec), - ledgerClient, - ScanAppClientConfig.DefaultScansRefreshInterval, - ScanAppClientConfig.DefaultAmuletRulesCacheTimeToLive, - upgradesConfig, - clock, - retryProvider, - loggerFactory, - )(ec, tc, mat, httpClient, templateJsonDecoder) + val future = for { + initialRound <- connection(SpliceLedgerConnectionPriority.Low) + .lookupUserMetadata( + config.ledgerApiUser, + BaseLedgerConnection.INITIAL_ROUND_USER_METADATA_KEY, + ) + bft <- BftScanConnection + .peerScanConnection( + () => + BftScanConnection.Bft.getPeerScansFromDsoRules( + dsoStore, + dsoStore.key.svParty, + )(tc, ec), + ledgerClient, + ScanAppClientConfig.DefaultScansRefreshInterval, + ScanAppClientConfig.DefaultAmuletRulesCacheTimeToLive, + upgradesConfig, + clock, + retryProvider, + loggerFactory, + initialRound = initialRound.map(_.toLong), + )(ec, tc, mat, httpClient, templateJsonDecoder) + } yield bft peerScanConnectionF = Some(future) future } diff --git a/project/ignore-patterns/canton_network_test_log.ignore.txt b/project/ignore-patterns/canton_network_test_log.ignore.txt index ab044cad67..cc23710368 100644 --- a/project/ignore-patterns/canton_network_test_log.ignore.txt +++ b/project/ignore-patterns/canton_network_test_log.ignore.txt @@ -173,6 +173,8 @@ api/sv/v0/onboard/validator \(POST\) resulted in a timeout # In BaseStorePerformanceTest, we run migrations in an unforked sbt JVM, so Flyway # finds the sbt test jar (apps-app_*-tests.jar). But, Flyway cannot open it, # so it skips it with this WARN. It contains no migrations, so this is harmless and unrelated to DB migrations. -# The same warning is emitted by both the FlywayExecutor and the ClassPathScanner loggers. -# a single pattern matching the message (the unloadable sbt tests jar) covers both. -Skipping unloadable jar file: file:.*-tests\.jar +Skipping unloadable jar file:.*FlywayExecutor + +# The BFT bootstrap test adds a dummy 5th SV with an unreachable scan URL (http://localhost:1) +# to raise f from 0 to 1. All BFT peers attempt to connect to this URL and log warnings. +Failed to connect to scan of.*NonZeroRoundBootstrapBftTimeBasedIntegrationTest diff --git a/test-full-class-names-sim-time.log b/test-full-class-names-sim-time.log index 6007a6c429..3432f76200 100644 --- a/test-full-class-names-sim-time.log +++ b/test-full-class-names-sim-time.log @@ -1,6 +1,7 @@ org.lfdecentralizedtrust.splice.integration.tests.DisabledWalletTimeBasedIntegrationTest org.lfdecentralizedtrust.splice.integration.tests.ExternallySignedTxsTimeBasedIntegrationTest org.lfdecentralizedtrust.splice.integration.tests.FollowAmuletConversionRateFeedTimeBasedIntegrationTest +org.lfdecentralizedtrust.splice.integration.tests.NonZeroRoundBootstrapBftTimeBasedIntegrationTest org.lfdecentralizedtrust.splice.integration.tests.ScanWithGradualStartsTimeBasedIntegrationTest org.lfdecentralizedtrust.splice.integration.tests.SvExpiredRewardsCollectionTimeBasedIntegrationTest org.lfdecentralizedtrust.splice.integration.tests.SvTimeBasedAmuletPriceIntegrationTest