From 926eddcd2a1720caa2ebcd19fbe999c46e23e2e0 Mon Sep 17 00:00:00 2001 From: Ilya Silvestrov Date: Tue, 30 Jun 2026 01:19:10 +0000 Subject: [PATCH 1/4] Log consensus scan URIs, revert BFT logging Log the scan URIs that formed the BFT consensus for the two reward-accounting reads. Consensus URIs are the `f + 1` URIs whose agreement formed the consensus, not every URI that happened to agree with it. - Expose consensus scan URIs via `getRewardAccountingRootHashWithScanUris` and `getRewardAccountingActivityTotalsWithScanUris` return values, and include them in the log messages of `CalculateRewardsTrigger` and `SummarizingMiningRoundTrigger` respectively. - Thread the consensus URIs through the BFT machinery: `executeCall` and the new private `bftCallWithScanUris` now return `(response, List[Uri])`; the `*WithScanUris` methods surface it while `bftCall` drops it to preserve the existing callers' signatures. - Revert the per-call logging customization introduced in #5768, keeping only the metrics collection and message formatting. All BFT disagreements are now logged uniformly at INFO including those for `getRewardAccountingRootHash` and `getRewardAccountingActivityTotals`. - Drop `ConsensusLogConfig` since disagreement logging no longer varies per BFT operation. - Update relevant tests. Tested locally: `TrafficBasedRewardsSvAppTimeBasedIntegrationTest` and `BftScanConnectionTest` pass. Signed-off-by: Ilya Silvestrov --- .../admin/api/client/BftScanConnection.scala | 131 ++++++++---------- .../api/client/BftScanConnectionTest.scala | 91 ++++++------ .../CalculateRewardsTrigger.scala | 8 +- .../SummarizingMiningRoundTrigger.scala | 6 +- 4 files changed, 109 insertions(+), 127 deletions(-) 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..5ec02cbece 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,25 @@ 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) + .map(_._1) + + private def bftCallWithScanUris[T]( + call: SingleScanConnection => Future[T], + endpoint: String, + callConfig: BftCallConfig, + consensusFailureLogLevel: Level = Level.WARN, + 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 +957,6 @@ class BftScanConnection( nTargetSuccess = callConfig.targetSuccess, logger, shortenResponsesForLog, - consensusLogConfig, connectionMetrics, ), logger, @@ -1010,15 +1021,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 +1047,13 @@ class BftScanConnection( }, endpoint = "getRewardAccountingActivityTotals", callConfig = callConfig, - consensusLogConfig = BftScanConnection.ConsensusLogConfig( - disagreementLogLevel = Level.WARN, - onlyLogDisagreementsInSuccessResponse = true, - agreementLogLevel = Some(Level.INFO), - ), ) - .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 +1069,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 +1094,22 @@ class BftScanConnection( }, endpoint = "getRewardAccountingRootHash", callConfig = callConfig, - consensusLogConfig = BftScanConnection.ConsensusLogConfig( - disagreementLogLevel = Level.WARN, - onlyLogDisagreementsInSuccessResponse = true, - agreementLogLevel = Some(Level.INFO), - ), ) - .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 +1146,18 @@ object BftScanConnection { nTargetSuccess: Int, logger: TracedLogger, shortenResponsesForLog: T => Any = identity[T], - consensusLogConfig: ConsensusLogConfig = ConsensusLogConfig(), 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 +1176,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 +1191,8 @@ object BftScanConnection { case Some(consensusResponse) => logDisagreements( logger, - consensusResponse, + consensusResponse.map(_._1), responses, - consensusLogConfig, connectionMetrics, ) } @@ -1226,10 +1238,8 @@ object BftScanConnection { logger: TracedLogger, consensusResponse: Try[T], responses: ConcurrentHashMap[BftScanConnection.ScanResponse[T], List[Uri]], - consensusLogConfig: ConsensusLogConfig, connectionMetrics: Option[ScanConnectionMetrics], )(implicit ec: ExecutionContext, tc: TraceContext, mc: MetricsContext): Unit = { - implicit val elc: ErrorLoggingContext = ErrorLoggingContext.fromTracedLogger(logger) def recordConsensus(url: Uri, consensus: String, extraLabels: Map[String, String]): Unit = connectionMetrics.foreach { metrics => val context = mc.merge( @@ -1256,28 +1266,15 @@ 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, - ) - } + logger.info( + s"""The following Scan URLs disagreed with consensus: + |${scanUrls.map(url => s" $url").mkString("\n")} + |consensus response: $consensusResponse + |disagreeing response: $disagreeingResponse""".stripMargin, + ) } } } @@ -2135,12 +2132,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..6df0968932 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 { @@ -1325,7 +1319,7 @@ class BftScanConnectionTest } } - "logs disagreements at WARN level" in { + "logs disagreements at INFO level" in { val round = 42L val connections = getMockedConnections(n = 4) makeMockReturnRootHashOk(connections(0), round, "aabb") @@ -1335,11 +1329,11 @@ class BftScanConnectionTest val bft = getBft(connections) loggerFactory - .assertEventuallyLogsSeq(SuppressionRule.Level(Level.WARN))( + .assertEventuallyLogsSeq(SuppressionRule.Level(Level.INFO))( bft.getRewardAccountingRootHash(round), logs => logs.exists(log => - log.level == Level.WARN && log.message.contains( + log.level == Level.INFO && log.message.contains( "disagreed with consensus" ) ) should be(true), @@ -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 { @@ -1449,7 +1438,7 @@ class BftScanConnectionTest } } - "logs disagreements at WARN level" in { + "logs disagreements at INFO level" in { val round = 42L val connections = getMockedConnections(n = 4) makeMockReturnActivityTotalsOk(connections(0), round, 100L, 10L, 5L) @@ -1459,11 +1448,11 @@ class BftScanConnectionTest val bft = getBft(connections) loggerFactory - .assertEventuallyLogsSeq(SuppressionRule.Level(Level.WARN))( + .assertEventuallyLogsSeq(SuppressionRule.Level(Level.INFO))( bft.getRewardAccountingActivityTotals(round), logs => logs.exists(log => - log.level == Level.WARN && log.message.contains( + log.level == Level.INFO && log.message.contains( "disagreed with consensus" ) ) should be(true), 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..e747e101e0 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 consensus 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..a46267fe1b 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 consensus scans: ${scanUris.mkString(", ")}.") ok case _ => totalsUnavailable("could not obtain reward accounting totals via BFT read.") } From 3ff698340382a622fc5c4c2730c447166aafb204 Mon Sep 17 00:00:00 2001 From: Ilya Silvestrov Date: Fri, 17 Jul 2026 09:13:15 +0000 Subject: [PATCH 2/4] Log reward-read BFT disagreements at WARN, test consensus URI logs - Re-introduce a plain `disagreementLogLevel` parameter in place of the dropped `ConsensusLogConfig`, threaded through `bftCallWithScanUris` -> `executeCall` -> `logDisagreements`. `getRewardAccountingRootHashWithScanUris` and `getRewardAccountingActivityTotalsWithScanUris` log disagreements at WARN. All other BFT calls keep the INFO default. Reaching agreement is still NOT logged separately exactly as it was before #5768. - Assert the consensus scan URI log messages in `TrafficBasedRewardsSvAppTimeBasedIntegrationTest`: the new `withExpectedRewardTriggersLogging` wrapper around `confirmBftRead` captures INFO logs of sv2's `CalculateRewardsTrigger` and `SummarizingMiningRoundTrigger` and checks both "... via BFT read from consensus scans: ." messages are emitted, with URIs restricted to sv1's and sv4's scans (sv3 is stopped and sv2's own scan is not part of its peer BFT connection). - Adjust the `BftScanConnectionTest` disagreement-level tests back to WARN for the two reward accounting reads. Tested locally: `TrafficBasedRewardsSvAppTimeBasedIntegrationTest` and `BftScanConnectionTest` pass. Signed-off-by: Ilya Silvestrov --- ...RewardsSvAppTimeBasedIntegrationTest.scala | 47 ++++++++++++++++++- .../admin/api/client/BftScanConnection.scala | 14 +++++- .../api/client/BftScanConnectionTest.scala | 12 ++--- 3 files changed, 64 insertions(+), 9 deletions(-) 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..7bbe95674b 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, @@ -347,11 +348,55 @@ class TrafficBasedRewardsSvAppTimeBasedIntegrationTest } } - confirmBftRead(bobParty) + withExpectedRewardTriggersLogging(_ => confirmBftRead(bobParty)) confirmMismatchingRootHashIsFlagged(bobParty) } + // sv2's CalculateRewardsTrigger and SummarizingMiningRoundTrigger report the + // scan URIs that formed the BFT consensus at INFO. Capture their logs for the + // whole duration of `body`. When `confirmBftRead` is passed as the body, BFT + // reads may start at any point after sv2's own scan starts answering + // CannotProvide. + // + // The oldest open round is read before running `body` and must match the + // round `body` performs the BFT reads for; this holds under sim time as + // rounds only advance when the test advances the clock. + private def withExpectedRewardTriggersLogging[A]( + body: Unit => A + )(implicit env: SpliceTestConsoleEnvironment): A = { + val round = oldestOpenRound + 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 consensus 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 should not be empty + forAll(scanUris)(uri => expectedScanUris should contain(uri)) + } + bftReadLogged("root-hash") + bftReadLogged("reward accounting totals") + }, + ) + } + private def metricValue( node: LocalInstanceReference, name: String, 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 5ec02cbece..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 @@ -907,7 +907,8 @@ class BftScanConnection( tc: TraceContext, ): Future[T] = bftCallWithScanUris( - call, endpoint, callConfig, consensusFailureLogLevel, shortenResponsesForLog) + call, endpoint, callConfig, consensusFailureLogLevel, + shortenResponsesForLog = shortenResponsesForLog) .map(_._1) private def bftCallWithScanUris[T]( @@ -915,6 +916,7 @@ class BftScanConnection( endpoint: String, callConfig: BftCallConfig, consensusFailureLogLevel: Level = Level.WARN, + disagreementLogLevel: Level = Level.INFO, shortenResponsesForLog: T => Any = identity[T], )(implicit ec: ExecutionContext, @@ -957,6 +959,7 @@ class BftScanConnection( nTargetSuccess = callConfig.targetSuccess, logger, shortenResponsesForLog, + disagreementLogLevel, connectionMetrics, ), logger, @@ -1047,6 +1050,7 @@ class BftScanConnection( }, endpoint = "getRewardAccountingActivityTotals", callConfig = callConfig, + disagreementLogLevel = Level.WARN, ) .transformWith { case Success((totals, consensusUris)) => @@ -1094,6 +1098,7 @@ class BftScanConnection( }, endpoint = "getRewardAccountingRootHash", callConfig = callConfig, + disagreementLogLevel = Level.WARN, ) .transformWith { case Success ((rootHash, consensusUris)) => @@ -1146,6 +1151,7 @@ object BftScanConnection { nTargetSuccess: Int, logger: TracedLogger, shortenResponsesForLog: T => Any = identity[T], + disagreementLogLevel: Level = Level.INFO, connectionMetrics: Option[ScanConnectionMetrics] = None, )(implicit ec: ExecutionContext, @@ -1193,6 +1199,7 @@ object BftScanConnection { logger, consensusResponse.map(_._1), responses, + disagreementLogLevel, connectionMetrics, ) } @@ -1238,8 +1245,10 @@ object BftScanConnection { logger: TracedLogger, consensusResponse: Try[T], responses: ConcurrentHashMap[BftScanConnection.ScanResponse[T], List[Uri]], + disagreementLogLevel: Level, connectionMetrics: Option[ScanConnectionMetrics], )(implicit ec: ExecutionContext, tc: TraceContext, mc: MetricsContext): Unit = { + implicit val elc: ErrorLoggingContext = ErrorLoggingContext.fromTracedLogger(logger) def recordConsensus(url: Uri, consensus: String, extraLabels: Map[String, String]): Unit = connectionMetrics.foreach { metrics => val context = mc.merge( @@ -1269,7 +1278,8 @@ object BftScanConnection { responses.forEach { (disagreeingResponse, scanUrls) => val extraLabels = disagreementLabels(disagreeingResponse) scanUrls.foreach(recordConsensus(_, "disagree", extraLabels)) - logger.info( + LoggerUtil.logAtLevel( + disagreementLogLevel, s"""The following Scan URLs disagreed with consensus: |${scanUrls.map(url => s" $url").mkString("\n")} |consensus response: $consensusResponse 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 6df0968932..7e74c2e54a 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 @@ -1319,7 +1319,7 @@ class BftScanConnectionTest } } - "logs disagreements at INFO level" in { + "logs disagreements at WARN level" in { val round = 42L val connections = getMockedConnections(n = 4) makeMockReturnRootHashOk(connections(0), round, "aabb") @@ -1329,11 +1329,11 @@ class BftScanConnectionTest val bft = getBft(connections) loggerFactory - .assertEventuallyLogsSeq(SuppressionRule.Level(Level.INFO))( + .assertEventuallyLogsSeq(SuppressionRule.Level(Level.WARN))( bft.getRewardAccountingRootHash(round), logs => logs.exists(log => - log.level == Level.INFO && log.message.contains( + log.level == Level.WARN && log.message.contains( "disagreed with consensus" ) ) should be(true), @@ -1438,7 +1438,7 @@ class BftScanConnectionTest } } - "logs disagreements at INFO level" in { + "logs disagreements at WARN level" in { val round = 42L val connections = getMockedConnections(n = 4) makeMockReturnActivityTotalsOk(connections(0), round, 100L, 10L, 5L) @@ -1448,11 +1448,11 @@ class BftScanConnectionTest val bft = getBft(connections) loggerFactory - .assertEventuallyLogsSeq(SuppressionRule.Level(Level.INFO))( + .assertEventuallyLogsSeq(SuppressionRule.Level(Level.WARN))( bft.getRewardAccountingActivityTotals(round), logs => logs.exists(log => - log.level == Level.INFO && log.message.contains( + log.level == Level.WARN && log.message.contains( "disagreed with consensus" ) ) should be(true), From 46aaf95c78e1e66f6a3612130451d47b37c82541 Mon Sep 17 00:00:00 2001 From: Ilya Silvestrov Date: Mon, 20 Jul 2026 21:43:57 +0000 Subject: [PATCH 3/4] Narrow 'withExpectedRewardTriggersLogging' use Use 'withExpectedRewardTriggersLogging' to check 'CalculateRewardsTrigger' and 'SummarizingMiningRoundTrigger' emit logs about BFT scans inside the 'confirmBftRead' call. Signed-off-by: Ilya Silvestrov --- ...RewardsSvAppTimeBasedIntegrationTest.scala | 191 +++++++++--------- 1 file changed, 94 insertions(+), 97 deletions(-) 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 7bbe95674b..3ae3b4d482 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 @@ -348,31 +348,26 @@ class TrafficBasedRewardsSvAppTimeBasedIntegrationTest } } - withExpectedRewardTriggersLogging(_ => confirmBftRead(bobParty)) + confirmBftRead(bobParty) confirmMismatchingRootHashIsFlagged(bobParty) } // sv2's CalculateRewardsTrigger and SummarizingMiningRoundTrigger report the - // scan URIs that formed the BFT consensus at INFO. Capture their logs for the - // whole duration of `body`. When `confirmBftRead` is passed as the body, BFT - // reads may start at any point after sv2's own scan starts answering - // CannotProvide. - // - // The oldest open round is read before running `body` and must match the - // round `body` performs the BFT reads for; this holds under sim time as - // rounds only advance when the test advances the clock. - private def withExpectedRewardTriggersLogging[A]( - body: Unit => A - )(implicit env: SpliceTestConsoleEnvironment): A = { - val round = oldestOpenRound + // 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(()), + 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. @@ -430,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 + } } } From d13b0857f72a8bf5d7be7acad5cf6a4b865b0c8f Mon Sep 17 00:00:00 2001 From: Ilya Silvestrov Date: Wed, 22 Jul 2026 23:24:37 +0000 Subject: [PATCH 4/4] Assert the exact expected scan URI count. Also, drop 'consensus' from BFT reward-read log messages Signed-off-by: Ilya Silvestrov --- .../TrafficBasedRewardsSvAppTimeBasedIntegrationTest.scala | 4 ++-- .../splice/scan/admin/api/client/BftScanConnectionTest.scala | 4 ++-- .../sv/automation/confirmation/CalculateRewardsTrigger.scala | 2 +- .../confirmation/SummarizingMiningRoundTrigger.scala | 2 +- 4 files changed, 6 insertions(+), 6 deletions(-) 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 3ae3b4d482..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 @@ -375,7 +375,7 @@ class TrafficBasedRewardsSvAppTimeBasedIntegrationTest def bftReadLogged(subject: String) = forAtLeast(1, logs) { log => val prefix = - s"Obtained the $subject for round $round via BFT read from consensus scans: " + 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 @@ -383,7 +383,7 @@ class TrafficBasedRewardsSvAppTimeBasedIntegrationTest .stripSuffix(".") .split(", ") .toSeq - scanUris should not be empty + scanUris.size should be(1) forAll(scanUris)(uri => expectedScanUris should contain(uri)) } bftReadLogged("root-hash") 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 7e74c2e54a..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 @@ -1259,7 +1259,7 @@ class BftScanConnectionTest case (GetRewardAccountingRootHashResponse.members.RewardAccountingRootHashOk(ok), uris) => ok.rootHash should be("aabb") ok.roundNumber should be(round) - uris.size should be >= 2 + uris.size should be(2) } } } @@ -1378,7 +1378,7 @@ class BftScanConnectionTest ok.totalAppActivityWeight should be(100L) ok.activePartiesCount should be(10L) ok.activityRecordsCount should be(5L) - uris.size should be >= 2 + uris.size should be(2) } } } 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 e747e101e0..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 @@ -163,7 +163,7 @@ abstract class CalculateRewardsTriggerBase( } yield response match { case (RewardAccountingRootHashOk(ok), scanUris) => logger.info( - s"Obtained the root-hash for round $round via BFT read from consensus scans: ${scanUris.mkString(", ")}." + 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 a46267fe1b..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 @@ -241,7 +241,7 @@ class SummarizingMiningRoundTrigger( response <- bftScan.getRewardAccountingActivityTotalsWithScanUris(round) } yield response match { case (RewardAccountingActivityTotalsOk(ok), scanUris) => - logger.info(s"Obtained the reward accounting totals for round $round via BFT read from consensus scans: ${scanUris.mkString(", ")}.") + 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.") }