SvApp: For TBA use total app activity in SummarizingMiningRoundTrigger

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

@@ -234,8 +234,12 @@ class TrafficBasedRewardsSvAppTimeBasedIntegrationTest
         val round = oldestOpenRound
         doTransfer(bobParty)
         // Need to advance by two rounds, see note below about last_archived_round
-        advanceRoundsToNextRoundOpening
-        advanceRoundsToNextRoundOpening
+        // 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
+        advanceTimeAndWaitForRoundOpening
 
         val (calculateRewardsCid, rootHash) =
           clue(
@@ -267,6 +271,10 @@ class TrafficBasedRewardsSvAppTimeBasedIntegrationTest
               .listConfirmations(startProcessingAction)
               .futureValue should have size 2
           }
+          sv1Backend.appState.dsoStore
+            .listOldestSummarizingMiningRounds()
+            .futureValue
+            .map(_.payload.round.number) should contain(round)
         }
 
         // This is trying to simulate AppActivityRecordMetaT's userVersion bump

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

@@ -172,34 +172,34 @@ abstract class TrafficBasedRewardsTimeBasedIntegrationTestBase
 
           // 3 initial advances to get open rounds with staggered opensAt
           for (round <- 1 to 3) {
-            advanceRoundsToNextRoundOpening
+            advanceTimeAndWaitForRoundOpening
             assertOldestOpenRound(round.toLong)
           }
 
           val id0 = settleTrade(aliceParty, bobParty, venueParty)
           grantFeaturedAppRight(splitwellWalletClient)
 
-          advanceRoundsToNextRoundOpening
+          advanceTimeAndWaitForRoundOpening
           assertOldestOpenRound(4)
 
           val id1 = settleTrade(aliceParty, bobParty, venueParty)
           grantFeaturedAppRight(aliceWalletClient)
 
-          advanceRoundsToNextRoundOpening
+          advanceTimeAndWaitForRoundOpening
           assertOldestOpenRound(5)
 
           settleTrade(aliceParty, bobParty, venueParty)
           settleTrade(aliceParty, bobParty, venueParty)
 
-          advanceRoundsToNextRoundOpening
+          advanceTimeAndWaitForRoundOpening
           assertOldestOpenRound(6)
 
           val id3 = settleTrade(aliceParty, bobParty, venueParty)
           settleTrade(aliceParty, bobParty, venueParty)
           settleTrade(aliceParty, bobParty, venueParty)
           settleTrade(aliceParty, bobParty, venueParty)
 
-          advanceRoundsToNextRoundOpening
+          advanceTimeAndWaitForRoundOpening
           assertOldestOpenRound(7)
 
           val id4 = settleTrade(aliceParty, bobParty, venueParty)
@@ -210,12 +210,12 @@ abstract class TrafficBasedRewardsTimeBasedIntegrationTestBase
           val (aliceCreateId, svExpireId) =
             aliceCreateAndSvExpireInstruction(aliceParty, bobParty)
 
-          advanceRoundsToNextRoundOpening
+          advanceTimeAndWaitForRoundOpening
           assertOldestOpenRound(8)
 
           // No activity for round 8
 
-          advanceRoundsToNextRoundOpening
+          advanceTimeAndWaitForRoundOpening
           assertOldestOpenRound(9)
 
           // Do only one DvP; this would not generate enough activity to reward the parties.
@@ -241,7 +241,7 @@ abstract class TrafficBasedRewardsTimeBasedIntegrationTestBase
             _ => sv1ScanBackend.lookupFeaturedAppRight(venueParty) shouldBe None,
           )
 
-          advanceRoundsToNextRoundOpening
+          advanceTimeAndWaitForRoundOpening
           assertOldestOpenRound(10)
 
           // Do five in a round to check nested BatchOfBatches processing
@@ -251,7 +251,7 @@ abstract class TrafficBasedRewardsTimeBasedIntegrationTestBase
           settleTrade(aliceParty, bobParty, venueParty)
           settleTrade(aliceParty, bobParty, venueParty)
 
-          advanceRoundsToNextRoundOpening
+          advanceTimeAndWaitForRoundOpening
           assertOldestOpenRound(11)
 
           val id7 = settleTrade(aliceParty, bobParty, venueParty)

apps/app/src/test/scala/org/lfdecentralizedtrust/splice/util/TimeTestUtil.scala

@@ -107,16 +107,47 @@ trait TimeTestUtil extends TestCommon {
         }
       }
     }
+    advanceTimeAndWaitForRoundAutomation(
+      durationToNextRoundOpening,
+      synchronizeExternalPartyConfigStates,
+    )
+  }
+
+  /** The amount of time to advance in order to reach the next mining round opening. */
+  private def durationToNextRoundOpening(implicit
+      env: SpliceTestConsoleEnvironment
+  ): Duration = {
     import math.Ordering.Implicits.*
     val now = sv1Backend.participantClient.ledger_api.time.get().toInstant
     val (openRounds, _) = sv1ScanBackend.getOpenAndIssuingMiningRounds()
-    val advanceWith = openRounds
+    openRounds
       .filter(round => now < round.contract.payload.opensAt)
       .map(_.contract.payload.opensAt)
       .minOption
       .map(minFutureOpen => Duration.between(now, minFutureOpen.plusSeconds(10)))
       .getOrElse(tickDurationWithBuffer)
-    advanceTimeAndWaitForRoundAutomation(advanceWith, synchronizeExternalPartyConfigStates)
+  }
+
+  /** Advance time, but only waits for the next round to open and does not wait
+    * for the summarizing and issuing round automation to catch up.
+    */
+  @nowarn("msg=match may not be exhaustive")
+  def advanceTimeAndWaitForRoundOpening(implicit env: SpliceTestConsoleEnvironment): Unit = {
+    val advanceWith = durationToNextRoundOpening
+    val (previousOpenRounds, _) = sv1ScanBackend.getOpenAndIssuingMiningRounds()
+    val Seq(lowestOpen, middleOpen, highestOpen) =
+      previousOpenRounds.map(_.contract.payload.round.number)
+
+    actAndCheck()("advancing time", advanceTime(advanceWith))(
+      s"waiting for open round automation (should create OpenMiningRound ${highestOpen + 1})",
+      _ => {
+        val (newOpenRounds, _) = sv1ScanBackend.getOpenAndIssuingMiningRounds()
+        val Seq(newLowestOpen, newMiddleOpen, newHighestOpen) =
+          newOpenRounds.map(_.contract.payload.round.number)
+        (newLowestOpen, newMiddleOpen, newHighestOpen) shouldBe
+          (lowestOpen + 1, middleOpen + 1, highestOpen + 1)
+      },
+    )
   }
 
   def advanceTimeAndUpdateExternalPartyConfigStates(implicit env: SpliceTestConsoleEnvironment) = {

apps/scan/src/main/openapi/scan.yaml

@@ -4208,6 +4208,10 @@ components:
         - total_app_activity_weight
         - active_parties_count
         - activity_records_count
+        - total_app_reward_minting_allowance
+        - total_app_reward_thresholded
+        - total_app_reward_unclaimed
+        - rewarded_app_provider_parties_count
       properties:
         status:
           type: string
@@ -4223,6 +4227,18 @@ components:
         activity_records_count:
           type: integer
           format: int64
+        total_app_reward_minting_allowance:
+          type: string
+          description: The total of all minting allowances granted to app providers in this round.
+        total_app_reward_thresholded:
+          type: string
+          description: Total amount of minting allowances that fell below the configured app reward threshold and was thus burned.
+        total_app_reward_unclaimed:
+          type: string
+          description: Total amount of app rewards which could not be attributed to app providers in this round because of limit on app rewards per activity (aka the app rewards cap). 
+        rewarded_app_provider_parties_count:
+          type: integer
+          format: int64
 
     RewardAccountingActivityTotalsUndetermined:
       type: object

apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/admin/api/client/BftScanConnection.scala

@@ -38,13 +38,16 @@ import org.lfdecentralizedtrust.splice.http.HttpClient
 import org.lfdecentralizedtrust.splice.http.v0.definitions.{
   AnsEntry,
   GetDsoInfoResponse,
+  GetRewardAccountingActivityTotalsResponse,
   GetRewardAccountingBatchResponse,
   GetRewardAccountingRootHashResponse,
   HoldingsSummaryRequestV1,
   HoldingsSummaryResponse,
   HoldingsSummaryResponseV1,
   LookupTransferCommandStatusResponse,
   MigrationSchedule,
+  RewardAccountingActivityTotalsOk,
+  RewardAccountingActivityTotalsUndetermined,
   RewardAccountingRootHashOk,
   RewardAccountingRootHashUndetermined,
 }
@@ -845,6 +848,54 @@ class BftScanConnection(
   ): Future[NonNegativeInt] =
     bftCall(_.getActivePhysicalSynchronizerSerial(), "getActivePhysicalSynchronizerSerial")
 
+  /** This is special because in addition to 'Ok' we can receive
+    * 'Undetermined' - This might indicate that scan is yet to process activity totals for this round
+    * 'CannotProvide' - Indicates that scan does not have required app-activity data to provide a response
+    *
+    * So simple equality comparison on responses is not possible, and we treat
+    * the two non-Ok responses as a "no response" by throwing IgnoreResponse so
+    * that this does not cause grouping in executeCall.
+    *
+    * And if no response could be obtained via bft we respond with 'Undetermined'
+    */
+  override def getRewardAccountingActivityTotals(roundNumber: Long)(implicit
+      ec: ExecutionContext,
+      tc: TraceContext,
+  ): Future[GetRewardAccountingActivityTotalsResponse] = {
+    val undetermined =
+      GetRewardAccountingActivityTotalsResponse(
+        RewardAccountingActivityTotalsUndetermined(status = "Undetermined")
+      )
+    val callConfig = BftCallConfig.default(scanList.scanConnections)
+    if (!callConfig.enoughAvailableScans) Future.successful(undetermined)
+    else
+      bftCall[RewardAccountingActivityTotalsOk](
+        call = scan =>
+          scan.getRewardAccountingActivityTotals(roundNumber).flatMap {
+            case GetRewardAccountingActivityTotalsResponse.members
+                  .RewardAccountingActivityTotalsOk(ok) =>
+              Future.successful(ok)
+            case _: GetRewardAccountingActivityTotalsResponse.members.RewardAccountingActivityTotalsUndetermined |
+                _: GetRewardAccountingActivityTotalsResponse.members.RewardAccountingActivityTotalsCannotProvide =>
+              Future.failed(BftScanConnection.IgnoreResponse(scan.url))
+          },
+        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)
+            )
+          )
+        )
+  }
+
   /** This is special because in addition to 'Ok' we can receive
     * 'Undetermined' - This might indicate that scan is yet to process root hash for this round
     * 'CannotProvide' - Indicates that scan does not have required app-activity data to provide a response

apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/admin/api/client/ScanConnection.scala

@@ -25,6 +25,7 @@ import org.lfdecentralizedtrust.splice.environment.*
 import org.lfdecentralizedtrust.splice.http.HttpClient
 import org.lfdecentralizedtrust.splice.http.v0.definitions.{
   GetDsoInfoResponse,
+  GetRewardAccountingActivityTotalsResponse,
   GetRewardAccountingBatchResponse,
   GetRewardAccountingRootHashResponse,
   HoldingsSummaryResponse,
@@ -336,6 +337,11 @@ trait ScanConnection
       tc: TraceContext,
   ): Future[NonNegativeInt]
 
+  def getRewardAccountingActivityTotals(roundNumber: Long)(implicit
+      ec: ExecutionContext,
+      tc: TraceContext,
+  ): Future[GetRewardAccountingActivityTotalsResponse]
+
   def getRewardAccountingRootHash(roundNumber: Long)(implicit
       ec: ExecutionContext,
       tc: TraceContext,

apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/admin/api/client/SingleScanConnection.scala

@@ -32,6 +32,7 @@ import org.lfdecentralizedtrust.splice.environment.{
 }
 import org.lfdecentralizedtrust.splice.http.HttpClient
 import org.lfdecentralizedtrust.splice.http.v0.definitions.{
+  GetRewardAccountingActivityTotalsResponse,
   GetRewardAccountingBatchResponse,
   GetRewardAccountingRootHashResponse,
   HoldingsSummaryRequestV1,
@@ -821,6 +822,15 @@ class SingleScanConnection private[client] (
       HttpScanAppClient.GetActivePhysicalSynchronizerSerial(),
     )
 
+  override def getRewardAccountingActivityTotals(roundNumber: Long)(implicit
+      ec: ExecutionContext,
+      tc: TraceContext,
+  ): Future[GetRewardAccountingActivityTotalsResponse] =
+    runHttpCmd(
+      config.adminApi.url,
+      HttpScanAppClient.GetRewardAccountingActivityTotals(roundNumber),
+    )
+
   override def getRewardAccountingRootHash(roundNumber: Long)(implicit
       ec: ExecutionContext,
       tc: TraceContext,