diff --git a/apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/tests/TrafficBasedRewardsSvAppTimeBasedIntegrationTest.scala b/apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/tests/TrafficBasedRewardsSvAppTimeBasedIntegrationTest.scala index b88a5c992d..8ef48d34ea 100644 --- a/apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/tests/TrafficBasedRewardsSvAppTimeBasedIntegrationTest.scala +++ b/apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/tests/TrafficBasedRewardsSvAppTimeBasedIntegrationTest.scala @@ -36,6 +36,7 @@ import org.lfdecentralizedtrust.splice.sv.automation.RewardMetricsTrigger import org.lfdecentralizedtrust.splice.sv.automation.confirmation.{ CalculateRewardsDryRunTrigger, CalculateRewardsTrigger, + SummarizingMiningRoundTrigger, } import org.lfdecentralizedtrust.splice.sv.automation.delegatebased.{ ProcessRewardsDryRunTrigger, @@ -352,6 +353,45 @@ class TrafficBasedRewardsSvAppTimeBasedIntegrationTest confirmMismatchingRootHashIsFlagged(bobParty) } + // sv2's CalculateRewardsTrigger and SummarizingMiningRoundTrigger report the + // scan URIs that formed the BFT consensus at INFO. This method captures the + // logs emitted while running the 'body' argument and asserts that sv2 + // obtained both the root-hash and the reward accounting totals for 'round' + // via BFT read from sv1 and sv4. + private def withExpectedRewardTriggersLogging[A](round: Long)( + body: => A + ): A = { + val bftReadLogs = + (SuppressionRule.forLogger[CalculateRewardsTrigger] || + SuppressionRule.forLogger[SummarizingMiningRoundTrigger]) && + SuppressionRule.LevelAndAbove(Level.INFO) + + loggerFactory.assertEventuallyLogsSeq(bftReadLogs)( + body, + logs => { + // sv3 is stopped and sv2's own scan is not part of its peer BFT connection, + // so only sv1's and sv4's scans can form the consensus. + val expectedScanUris = Set("http://localhost:5012", "http://localhost:5312") + def bftReadLogged(subject: String) = + forAtLeast(1, logs) { log => + val prefix = + s"Obtained the $subject for round $round via BFT read from scans: " + log.loggerName should include("SV=sv2") + log.message should include(prefix) + val scanUris = log.message + .substring(log.message.indexOf(prefix) + prefix.length) + .stripSuffix(".") + .split(", ") + .toSeq + scanUris.size should be(1) + forAll(scanUris)(uri => expectedScanUris should contain(uri)) + } + bftReadLogged("root-hash") + bftReadLogged("reward accounting totals") + }, + ) + } + private def metricValue( node: LocalInstanceReference, name: String, @@ -385,99 +425,101 @@ class TrafficBasedRewardsSvAppTimeBasedIntegrationTest // Pausing this ensures that the root-hash is not calculated while we advance round val sv2RewardComputation = sv2ScanBackend.automation.trigger[RewardComputationTrigger] - // Here we ensure that SV2 has done ingestion of app-activity for the round just closed - // But then its AppActivityRecordMetaT is bumped so that it cannot compute the - // root-hash for the round. - val (calculateRewardsCid, round) = setTriggersWithin( - triggersToPauseAtStart = Seq(sv2CalculateRewards, sv2RewardComputation) - ) { - val round = oldestOpenRound - doTransfer(bobParty) - // Note: we can't use advanceRoundsToNextRoundOpening here, as it blocks - // on summarizing and issuing round to complete, and here the - // summarizing round will block until the sv2 provides the round totals - // via bft read. - advanceTimeAndWaitForRoundOpening - - val (calculateRewardsCid, rootHash) = - clue( - s"Round $round just closed: its CalculateRewardsV2 exists and sv1 serves root-hash" - ) { - eventually() { - val calc = sv1Backend.appState.dsoStore - .listCalculateRewardsV2() - .futureValue - .filterNot(_.payload.dryRun) - .find(_.payload.round.number == round) - .value - val rootHash = inside(sv1ScanBackend.getRewardAccountingRootHash(round)) { - case GetRewardAccountingRootHashResponse.members.RewardAccountingRootHashOk(h) => - h.rootHash + val round = oldestOpenRound + withExpectedRewardTriggersLogging(round) { + // Here we ensure that SV2 has done ingestion of app-activity for the round just closed + // But then its AppActivityRecordMetaT is bumped so that it cannot compute the + // root-hash for the round. + val calculateRewardsCid = setTriggersWithin( + triggersToPauseAtStart = Seq(sv2CalculateRewards, sv2RewardComputation) + ) { + doTransfer(bobParty) + // Note: we can't use advanceRoundsToNextRoundOpening here, as it blocks + // on summarizing and issuing round to complete, and here the + // summarizing round will block until the sv2 provides the round totals + // via bft read. + advanceTimeAndWaitForRoundOpening + + val (calculateRewardsCid, rootHash) = + clue( + s"Round $round just closed: its CalculateRewardsV2 exists and sv1 serves root-hash" + ) { + eventually() { + val calc = sv1Backend.appState.dsoStore + .listCalculateRewardsV2() + .futureValue + .filterNot(_.payload.dryRun) + .find(_.payload.round.number == round) + .value + val rootHash = inside(sv1ScanBackend.getRewardAccountingRootHash(round)) { + case GetRewardAccountingRootHashResponse.members.RewardAccountingRootHashOk(h) => + h.rootHash + } + (calc.contractId, rootHash) } - (calc.contractId, rootHash) } - } - clue(s"Only sv1 and sv4 confirm round $round, so it is not yet processed") { - eventually() { - val startProcessingAction = new ARC_AmuletRules( - new CRARC_StartProcessingRewardsV2( - new AmuletRules_StartProcessingRewardsV2(calculateRewardsCid, new Hash(rootHash)) + clue(s"Only sv1 and sv4 confirm round $round, so it is not yet processed") { + eventually() { + val startProcessingAction = new ARC_AmuletRules( + new CRARC_StartProcessingRewardsV2( + new AmuletRules_StartProcessingRewardsV2(calculateRewardsCid, new Hash(rootHash)) + ) ) - ) + sv1Backend.appState.dsoStore + .listConfirmations(startProcessingAction) + .futureValue should have size 2 + } sv1Backend.appState.dsoStore - .listConfirmations(startProcessingAction) - .futureValue should have size 2 + .listOldestSummarizingMiningRounds() + .futureValue + .map(_.payload.round.number) should contain(round) } - sv1Backend.appState.dsoStore - .listOldestSummarizingMiningRounds() - .futureValue - .map(_.payload.round.number) should contain(round) - } - // This is trying to simulate AppActivityRecordMetaT's userVersion bump - // albeit in a direct way, to avoid restart of scan app, etc. - actAndCheck( - s"Reset sv2's earliest-ingested round to $round", { - val sv2Db = sv2ScanBackend.appState.storage match { - case db: DbStorage => db - case other => fail(s"Expected DbStorage") - } - implicit val closeContext: CloseContext = CloseContext(sv2Db) - sv2Db - .update_( - sqlu"""update app_activity_record_meta - set earliest_ingested_round = $round, - last_archived_round = null""", - "test.increaseAppActivityMeta_EarliestIngestedRound", - ) - .futureValueUS - }, - )( - s"sv2's own scan now answers CannotProvide for round $round", - _ => - sv2ScanBackend.getRewardAccountingRootHash(round) shouldBe - a[GetRewardAccountingRootHashResponse.members.RewardAccountingRootHashCannotProvide], - ) + // This is trying to simulate AppActivityRecordMetaT's userVersion bump + // albeit in a direct way, to avoid restart of scan app, etc. + actAndCheck( + s"Reset sv2's earliest-ingested round to $round", { + val sv2Db = sv2ScanBackend.appState.storage match { + case db: DbStorage => db + case other => fail(s"Expected DbStorage") + } + implicit val closeContext: CloseContext = CloseContext(sv2Db) + sv2Db + .update_( + sqlu"""update app_activity_record_meta + set earliest_ingested_round = $round, + last_archived_round = null""", + "test.increaseAppActivityMeta_EarliestIngestedRound", + ) + .futureValueUS + }, + )( + s"sv2's own scan now answers CannotProvide for round $round", + _ => + sv2ScanBackend.getRewardAccountingRootHash(round) shouldBe + a[GetRewardAccountingRootHashResponse.members.RewardAccountingRootHashCannotProvide], + ) - (calculateRewardsCid, round) - } + calculateRewardsCid + } - // setTriggersWithin has resumed sv2's CalculateRewardsTrigger. sv3 is stopped and sv2's own - // scan CannotProvide, so the deciding 3rd confirmation can only come from sv2 via bft read. - clue(s"sv2's own scan still answers CannotProvide for round $round") { - sv2ScanBackend.getRewardAccountingRootHash(round) shouldBe - a[GetRewardAccountingRootHashResponse.members.RewardAccountingRootHashCannotProvide] - } + // setTriggersWithin has resumed sv2's CalculateRewardsTrigger. sv3 is stopped and sv2's own + // scan CannotProvide, so the deciding 3rd confirmation can only come from sv2 via bft read. + clue(s"sv2's own scan still answers CannotProvide for round $round") { + sv2ScanBackend.getRewardAccountingRootHash(round) shouldBe + a[GetRewardAccountingRootHashResponse.members.RewardAccountingRootHashCannotProvide] + } - clue( - s"sv2 reads round $round from the sv1 and sv4, and supplies the 3rd confirmation vote" - ) { - eventually() { - sv1Backend.appState.dsoStore - .listCalculateRewardsV2() - .futureValue - .map(_.contractId) should not contain calculateRewardsCid + clue( + s"sv2 reads round $round from the sv1 and sv4, and supplies the 3rd confirmation vote" + ) { + eventually() { + sv1Backend.appState.dsoStore + .listCalculateRewardsV2() + .futureValue + .map(_.contractId) should not contain calculateRewardsCid + } } } 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..7ac4c48453 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 @@ -901,13 +901,27 @@ class BftScanConnection( endpoint: String, callConfig: BftCallConfig = BftCallConfig.default(scanList.scanConnections), consensusFailureLogLevel: Level = Level.WARN, - consensusLogConfig: BftScanConnection.ConsensusLogConfig = - BftScanConnection.ConsensusLogConfig(), shortenResponsesForLog: T => Any = identity[T], )(implicit ec: ExecutionContext, tc: TraceContext, - ): Future[T] = { + ): Future[T] + = bftCallWithScanUris( + call, endpoint, callConfig, consensusFailureLogLevel, + shortenResponsesForLog = shortenResponsesForLog) + .map(_._1) + + private def bftCallWithScanUris[T]( + call: SingleScanConnection => Future[T], + endpoint: String, + callConfig: BftCallConfig, + consensusFailureLogLevel: Level = Level.WARN, + disagreementLogLevel: Level = Level.INFO, + shortenResponsesForLog: T => Any = identity[T], + )(implicit + ec: ExecutionContext, + tc: TraceContext, + ): Future[(T, List[Uri])] = { implicit val mc: MetricsContext = MetricsContext("request" -> endpoint) val connections = scanList.scanConnections @@ -945,7 +959,7 @@ class BftScanConnection( nTargetSuccess = callConfig.targetSuccess, logger, shortenResponsesForLog, - consensusLogConfig, + disagreementLogLevel, connectionMetrics, ), logger, @@ -1010,15 +1024,21 @@ class BftScanConnection( override def getRewardAccountingActivityTotals(roundNumber: Long)(implicit ec: ExecutionContext, tc: TraceContext, - ): Future[GetRewardAccountingActivityTotalsResponse] = { + ): Future[GetRewardAccountingActivityTotalsResponse] + = getRewardAccountingActivityTotalsWithScanUris(roundNumber).map(_._1) + + def getRewardAccountingActivityTotalsWithScanUris(roundNumber: Long)(implicit + ec: ExecutionContext, + tc: TraceContext, + ): Future[(GetRewardAccountingActivityTotalsResponse, List[Uri])] = { val undetermined = GetRewardAccountingActivityTotalsResponse( RewardAccountingActivityTotalsUndetermined(status = "Undetermined") ) val callConfig = BftCallConfig.default(scanList.scanConnections) - if (!callConfig.enoughAvailableScans) Future.successful(undetermined) + if (!callConfig.enoughAvailableScans) Future.successful((undetermined, Nil)) else - bftCall[RewardAccountingActivityTotalsOk]( + bftCallWithScanUris[RewardAccountingActivityTotalsOk]( call = scan => scan.getRewardAccountingActivityTotals(roundNumber).flatMap { case GetRewardAccountingActivityTotalsResponse.members @@ -1030,19 +1050,14 @@ class BftScanConnection( }, endpoint = "getRewardAccountingActivityTotals", callConfig = callConfig, - consensusLogConfig = BftScanConnection.ConsensusLogConfig( - disagreementLogLevel = Level.WARN, - onlyLogDisagreementsInSuccessResponse = true, - agreementLogLevel = Some(Level.INFO), - ), + disagreementLogLevel = Level.WARN, ) - .transform(tryTotals => - Success( - tryTotals.toOption.fold(undetermined)(ok => - GetRewardAccountingActivityTotalsResponse(ok) - ) - ) - ) + .transformWith { + case Success((totals, consensusUris)) => + Future.successful( + ( GetRewardAccountingActivityTotalsResponse(totals), consensusUris) ) + case Failure(_) => Future.successful((undetermined, Nil)) + } } /** This is special because in addition to 'Ok' we can receive @@ -1058,15 +1073,21 @@ class BftScanConnection( override def getRewardAccountingRootHash(roundNumber: Long)(implicit ec: ExecutionContext, tc: TraceContext, - ): Future[GetRewardAccountingRootHashResponse] = { + ): Future[GetRewardAccountingRootHashResponse] + = getRewardAccountingRootHashWithScanUris(roundNumber).map(_._1) + + def getRewardAccountingRootHashWithScanUris(roundNumber: Long)(implicit + ec: ExecutionContext, + tc: TraceContext, + ): Future[(GetRewardAccountingRootHashResponse, List[Uri])] = { val undetermined = GetRewardAccountingRootHashResponse( RewardAccountingRootHashUndetermined(status = "Undetermined") ) val callConfig = BftCallConfig.default(scanList.scanConnections) - if (!callConfig.enoughAvailableScans) Future.successful(undetermined) + if (!callConfig.enoughAvailableScans) Future.successful((undetermined, Nil)) else - bftCall[String]( + bftCallWithScanUris[String]( call = scan => scan.getRewardAccountingRootHash(roundNumber).flatMap { case GetRewardAccountingRootHashResponse.members.RewardAccountingRootHashOk(ok) => @@ -1077,25 +1098,23 @@ class BftScanConnection( }, endpoint = "getRewardAccountingRootHash", callConfig = callConfig, - consensusLogConfig = BftScanConnection.ConsensusLogConfig( - disagreementLogLevel = Level.WARN, - onlyLogDisagreementsInSuccessResponse = true, - agreementLogLevel = Some(Level.INFO), - ), + disagreementLogLevel = Level.WARN, ) - .transform(tryRootHash => - Success( - tryRootHash.toOption.fold(undetermined)(rootHash => - GetRewardAccountingRootHashResponse( - RewardAccountingRootHashOk( - status = "Ok", - roundNumber = roundNumber, - rootHash = rootHash, + .transformWith { + case Success ((rootHash, consensusUris)) => + Future.successful( + ( GetRewardAccountingRootHashResponse( + RewardAccountingRootHashOk( + status = "Ok", + roundNumber = roundNumber, + rootHash = rootHash, + ) ) + , consensusUris.map(_.toString) ) ) - ) - ) + case Failure(_) => Future.successful((undetermined, Nil)) + } } /** The batch contents are verifiable via the hash, so BFT agreement across scans is not @@ -1132,19 +1151,19 @@ object BftScanConnection { nTargetSuccess: Int, logger: TracedLogger, shortenResponsesForLog: T => Any = identity[T], - consensusLogConfig: ConsensusLogConfig = ConsensusLogConfig(), + disagreementLogLevel: Level = Level.INFO, connectionMetrics: Option[ScanConnectionMetrics] = None, )(implicit ec: ExecutionContext, tc: TraceContext, mc: MetricsContext = MetricsContext.Empty, - ): Future[T] = { + ): Future[(T, List[Uri])] = { require(requestFrom.nonEmpty, "At least one request must be made.") val responses = new ConcurrentHashMap[BftScanConnection.ScanResponse[T], List[Uri]]() val nResponsesDone = new AtomicInteger(0) - val finalResponse = Promise[T]() + val finalResponse = Promise[(T, List[Uri])]() requestFrom.foreach { scan => call(scan) @@ -1163,7 +1182,7 @@ object BftScanConnection { case _ => true } if (considerResponseForQuorum && agreements.size == nTargetSuccess) { // consensus has been reached - finalResponse.tryComplete(response): Unit + finalResponse.tryComplete(response.map(r => (r, agreements))): Unit } if (nResponsesDone.incrementAndGet() == requestFrom.size) { // all Scans are done @@ -1178,9 +1197,9 @@ object BftScanConnection { case Some(consensusResponse) => logDisagreements( logger, - consensusResponse, + consensusResponse.map(_._1), responses, - consensusLogConfig, + disagreementLogLevel, connectionMetrics, ) } @@ -1226,7 +1245,7 @@ object BftScanConnection { logger: TracedLogger, consensusResponse: Try[T], responses: ConcurrentHashMap[BftScanConnection.ScanResponse[T], List[Uri]], - consensusLogConfig: ConsensusLogConfig, + disagreementLogLevel: Level, connectionMetrics: Option[ScanConnectionMetrics], )(implicit ec: ExecutionContext, tc: TraceContext, mc: MetricsContext): Unit = { implicit val elc: ErrorLoggingContext = ErrorLoggingContext.fromTracedLogger(logger) @@ -1256,28 +1275,16 @@ object BftScanConnection { keyToGroupResponses(consensusResponse).foreach { consensusResponseKey => val agreeingScanUrls = responses.remove(consensusResponseKey) agreeingScanUrls.foreach(recordConsensus(_, "agree", Map.empty)) - consensusLogConfig.agreementLogLevel.foreach { level => - LoggerUtil.logAtLevel( - level, - s"Reached consensus from:\n${agreeingScanUrls.mkString("\n")}", - ) - } responses.forEach { (disagreeingResponse, scanUrls) => val extraLabels = disagreementLabels(disagreeingResponse) scanUrls.foreach(recordConsensus(_, "disagree", extraLabels)) - val shouldLog = disagreeingResponse match { - case _: SuccessfulResponse[?] => true - case _ => !consensusLogConfig.onlyLogDisagreementsInSuccessResponse - } - if (shouldLog) { - LoggerUtil.logAtLevel( - consensusLogConfig.disagreementLogLevel, - s"""The following Scan URLs disagreed with consensus: - |${scanUrls.map(url => s" $url").mkString("\n")} - |consensus response: $consensusResponse - |disagreeing response: $disagreeingResponse""".stripMargin, - ) - } + LoggerUtil.logAtLevel( + disagreementLogLevel, + s"""The following Scan URLs disagreed with consensus: + |${scanUrls.map(url => s" $url").mkString("\n")} + |consensus response: $consensusResponse + |disagreeing response: $disagreeingResponse""".stripMargin, + ) } } } @@ -2135,12 +2142,6 @@ object BftScanConnection { extends RuntimeException(s"Scan $url has no answer to contribute to consensus") with NoStackTrace - case class ConsensusLogConfig( - disagreementLogLevel: Level = Level.INFO, - onlyLogDisagreementsInSuccessResponse: Boolean = false, - agreementLogLevel: Option[Level] = None, - ) - private sealed trait ScanResponse[+T] private case class SuccessfulResponse[+T](response: T) extends ScanResponse[T] private case class HttpFailureResponse[+T](status: StatusCode, body: Json) extends ScanResponse[T] 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..72515e9d9e 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 @@ -97,6 +97,7 @@ class BftScanConnectionTest when(m.config).thenReturn( ScanAppClientConfig(NetworkAppClientConfig(scanUrl(n))) ) + when(m.url).thenReturn(Uri(scanUrl(n))) m } connections.foreach { connection => @@ -1043,7 +1044,7 @@ class BftScanConnectionTest connections.tail.foreach(c => when(c.getDsoPartyId()).thenReturn(delayedSuccess)) for { - result <- BftScanConnection.executeCall(call, connections, nTargetSuccess = 1, logger) + (result, _) <- BftScanConnection.executeCall(call, connections, nTargetSuccess = 1, logger) } yield result should be(partyIdA) } @@ -1111,7 +1112,7 @@ class BftScanConnectionTest } for { - result <- BftScanConnection.executeCall( + (result, _) <- BftScanConnection.executeCall( call, connections, nTargetSuccess = 2, @@ -1158,7 +1159,7 @@ class BftScanConnectionTest makeMockFail(connections(2), notFoundFailure) for { - result <- BftScanConnection.executeCall( + (result, _) <- BftScanConnection.executeCall( call, connections, nTargetSuccess = 2, @@ -1194,7 +1195,7 @@ class BftScanConnectionTest } for { - result <- BftScanConnection.executeCall( + (result, _) <- BftScanConnection.executeCall( call, connections, nTargetSuccess = 2, @@ -1245,29 +1246,22 @@ class BftScanConnectionTest // With n=4, we query only two connections randomly, and even with // retries a single call can fail to reach consensus. - def attempt(remaining: Int): Future[GetRewardAccountingRootHashResponse] = - bft.getRewardAccountingRootHash(round).flatMap { - case ok: GetRewardAccountingRootHashResponse.members.RewardAccountingRootHashOk => - Future.successful(ok) + def attempt(remaining: Int): Future[(GetRewardAccountingRootHashResponse, List[Uri])] = + bft.getRewardAccountingRootHashWithScanUris(round).flatMap { + case (ok : GetRewardAccountingRootHashResponse.members.RewardAccountingRootHashOk, uris) => + Future.successful((ok, uris)) case _ if remaining > 1 => attempt(remaining - 1) case other => Future.successful(other) } - loggerFactory - .assertEventuallyLogsSeq(SuppressionRule.LevelAndAbove(Level.INFO))( - attempt(100).map { resp => - inside(resp) { - case GetRewardAccountingRootHashResponse.members.RewardAccountingRootHashOk(ok) => - ok.rootHash should be("aabb") - ok.roundNumber should be(round) - } - }, - logs => - logs.exists(l => - l.level == Level.INFO && l.message.contains("Reached consensus from") - ) should be(true), - ) - .map(_ => succeed) + attempt(100).map { resp => + inside(resp) { + case (GetRewardAccountingRootHashResponse.members.RewardAccountingRootHashOk(ok), uris) => + ok.rootHash should be("aabb") + ok.roundNumber should be(round) + uris.size should be(2) + } + } } "returns Undetermined when no quorum agrees on a hash" in { @@ -1366,32 +1360,27 @@ class BftScanConnectionTest // With n=4, we query only two connections randomly, and even with // retries a single call can fail to reach consensus. - def attempt(remaining: Int): Future[GetRewardAccountingActivityTotalsResponse] = - bft.getRewardAccountingActivityTotals(round).flatMap { - case ok: GetRewardAccountingActivityTotalsResponse.members.RewardAccountingActivityTotalsOk => - Future.successful(ok) + def attempt(remaining: Int): Future[(GetRewardAccountingActivityTotalsResponse, List[Uri])] = + bft.getRewardAccountingActivityTotalsWithScanUris(round).flatMap { + case (ok: GetRewardAccountingActivityTotalsResponse.members.RewardAccountingActivityTotalsOk, uris) => + Future.successful((ok, uris)) case _ if remaining > 1 => attempt(remaining - 1) case other => Future.successful(other) } - loggerFactory - .assertEventuallyLogsSeq(SuppressionRule.LevelAndAbove(Level.INFO))( - attempt(100).map { resp => - inside(resp) { - case GetRewardAccountingActivityTotalsResponse.members - .RewardAccountingActivityTotalsOk(ok) => - ok.roundNumber should be(round) - ok.totalAppActivityWeight should be(100L) - ok.activePartiesCount should be(10L) - ok.activityRecordsCount should be(5L) - } - }, - logs => - logs.exists(l => - l.level == Level.INFO && l.message.contains("Reached consensus from") - ) should be(true), - ) - .map(_ => succeed) + attempt(100).map { resp => + inside(resp) { + case ( GetRewardAccountingActivityTotalsResponse.members + .RewardAccountingActivityTotalsOk(ok) + , uris + ) => + ok.roundNumber should be(round) + ok.totalAppActivityWeight should be(100L) + ok.activePartiesCount should be(10L) + ok.activityRecordsCount should be(5L) + uris.size should be(2) + } + } } "returns Undetermined when no quorum agrees on the totals" in { diff --git a/apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/automation/confirmation/CalculateRewardsTrigger.scala b/apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/automation/confirmation/CalculateRewardsTrigger.scala index 6682db3c36..ff1c6dd714 100644 --- a/apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/automation/confirmation/CalculateRewardsTrigger.scala +++ b/apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/automation/confirmation/CalculateRewardsTrigger.scala @@ -159,10 +159,12 @@ abstract class CalculateRewardsTriggerBase( rewardMetrics.calculateRewardsRootHashBftReads.mark() for { bftScan <- getPeerBftScanConnection() - response <- bftScan.getRewardAccountingRootHash(round) + response <- bftScan.getRewardAccountingRootHashWithScanUris(round) } yield response match { - case RewardAccountingRootHashOk(ok) => - logger.info(s"Obtained the root-hash for round $round via BFT read.") + case (RewardAccountingRootHashOk(ok), scanUris) => + logger.info( + s"Obtained the root-hash for round $round via BFT read from scans: ${scanUris.mkString(", ")}." + ) new Hash(ok.rootHash) case _ => rootHashUnavailable("could not obtain root-hash via BFT read.") } diff --git a/apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/automation/confirmation/SummarizingMiningRoundTrigger.scala b/apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/automation/confirmation/SummarizingMiningRoundTrigger.scala index 98ab463033..4d4e943ada 100644 --- a/apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/automation/confirmation/SummarizingMiningRoundTrigger.scala +++ b/apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/automation/confirmation/SummarizingMiningRoundTrigger.scala @@ -238,10 +238,10 @@ class SummarizingMiningRoundTrigger( miningRoundMetrics.summarizingRoundTotalsBftReads.mark() for { bftScan <- bftScanConnectionF() - response <- bftScan.getRewardAccountingActivityTotals(round) + response <- bftScan.getRewardAccountingActivityTotalsWithScanUris(round) } yield response match { - case RewardAccountingActivityTotalsOk(ok) => - logger.info(s"Obtained the reward accounting totals for round $round via BFT read.") + case (RewardAccountingActivityTotalsOk(ok), scanUris) => + logger.info(s"Obtained the reward accounting totals for round $round via BFT read from scans: ${scanUris.mkString(", ")}.") ok case _ => totalsUnavailable("could not obtain reward accounting totals via BFT read.") }