Fix getHoldingsSummary returning bad counts when too many contracts (#1072)

Signed-off-by: Oriol Muñoz <oriol.munoz@digitalasset.com>

Author: OriolMunoz-da

apps/common/src/main/scala/org/lfdecentralizedtrust/splice/store/Limit.scala

@@ -89,6 +89,28 @@ trait LimitHelpers { _: NamedLogging =>
     }
   }
 
+  protected final def applyLimitOrFail[CC[_], C](
+      name: String,
+      limit: Limit,
+      result: C & scala.collection.IterableOps[?, CC, C],
+  ): C = {
+    limit match {
+      case PageLimit(limit) =>
+        result.take(limit.intValue())
+      case HardLimit(limit) =>
+        val resultSize = result.size
+        if (resultSize > limit) {
+          throw io.grpc.Status.FAILED_PRECONDITION
+            .withDescription(
+              s"Size of the result exceeded the limit in $name. Result size: ${resultSize.toLong}. Limit: ${limit.toLong}"
+            )
+            .asRuntimeException()
+        } else {
+          result
+        }
+    }
+  }
+
   protected def sqlLimit(limit: Limit): Int = {
     limit match {
       case HardLimit(limit) => limit + 1

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

@@ -2619,6 +2619,7 @@ components:
           type: array
           items:
             type: string
+          minItems: 1
           description: |
             Filters by contracts in which these party_ids are the owners of the amulets.
 
@@ -2644,6 +2645,7 @@ components:
           type: array
           items:
             type: string
+          minItems: 1
           description: |
             The owners for which to compute the summary.
         as_of_round:

apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/admin/http/HttpScanHandler.scala

@@ -3,7 +3,7 @@
 
 package org.lfdecentralizedtrust.splice.scan.admin.http
 
-import cats.data.OptionT
+import cats.data.{NonEmptyVector, OptionT}
 import cats.implicits.toTraverseOps
 import cats.syntax.either.*
 import cats.syntax.traverseFilter.*
@@ -1361,7 +1361,7 @@ class HttpScanHandler(
               CantonTimestamp.assertFromInstant(recordTime.toInstant),
               after,
               PageLimit.tryCreate(pageSize),
-              ownerPartyIds.map(PartyId.tryFromProtoPrimitive),
+              nonEmptyOrFail("ownerPartyIds", ownerPartyIds).map(PartyId.tryFromProtoPrimitive),
             )
             .map { result =>
               ScanResource.GetHoldingsStateAtResponseOK(
@@ -1410,7 +1410,7 @@ class HttpScanHandler(
               .getHoldingsSummary(
                 migrationId,
                 CantonTimestamp.assertFromInstant(recordTime.toInstant),
-                partyIds.map(PartyId.tryFromProtoPrimitive),
+                nonEmptyOrFail("partyIds", partyIds).map(PartyId.tryFromProtoPrimitive),
                 round,
               )
           } yield ScanResource.GetHoldingsSummaryAtResponse.OK(
@@ -1438,6 +1438,18 @@ class HttpScanHandler(
     }
   }
 
+  private def nonEmptyOrFail[A](fieldName: String, vec: Vector[A]): NonEmptyVector[A] = {
+    NonEmptyVector
+      .fromVector(vec)
+      .getOrElse(
+        throw io.grpc.Status.INVALID_ARGUMENT
+          .withDescription(
+            s"Expected '$fieldName' to contain at least one item, but contained none."
+          )
+          .asRuntimeException()
+      )
+  }
+
   override def getAggregatedRounds(respond: ScanResource.GetAggregatedRoundsResponse.type)()(
       extracted: com.digitalasset.canton.tracing.TraceContext
   ): Future[ScanResource.GetAggregatedRoundsResponse] = {

apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/AcsSnapshotStore.scala

@@ -3,6 +3,7 @@
 
 package org.lfdecentralizedtrust.splice.scan.store
 
+import cats.data.NonEmptyVector
 import com.daml.ledger.javaapi.data.CreatedEvent
 import org.lfdecentralizedtrust.splice.codegen.java.splice.amulet.{Amulet, LockedAmulet}
 import org.lfdecentralizedtrust.splice.scan.store.AcsSnapshotStore.{
@@ -240,7 +241,8 @@ class AcsSnapshotStore(
           "queryAcsSnapshot.getCreatedEvents",
         )
     } yield {
-      val eventsInPage = applyLimit("queryAcsSnapshot", limit, events.map(_._2.toCreatedEvent))
+      val eventsInPage =
+        applyLimitOrFail("queryAcsSnapshot", limit, events.map(_._2.toCreatedEvent))
       val afterToken = if (eventsInPage.size == limit.limit) events.lastOption.map(_._1) else None
       QueryAcsSnapshotResult(
         migrationId = migrationId,
@@ -256,18 +258,19 @@ class AcsSnapshotStore(
       snapshot: CantonTimestamp,
       after: Option[Long],
       limit: Limit,
-      partyIds: Seq[PartyId],
+      partyIds: NonEmptyVector[PartyId],
   )(implicit tc: TraceContext): Future[QueryAcsSnapshotResult] = {
     this
       .queryAcsSnapshot(
         migrationId,
         snapshot,
         after,
         limit,
-        partyIds,
+        partyIds.toVector,
         AcsSnapshotStore.holdingsTemplates,
       )
       .map { result =>
+        val partyIdsSet = partyIds.toVector.toSet
         QueryAcsSnapshotResult(
           result.migrationId,
           result.snapshotRecordTime,
@@ -277,8 +280,10 @@ class AcsSnapshotStore(
                 .decodeHoldingContract(createdEvent.event)
                 .fold(
                   locked =>
-                    partyIds.contains(PartyId.tryFromProtoPrimitive(locked.payload.amulet.owner)),
-                  amulet => partyIds.contains(PartyId.tryFromProtoPrimitive(amulet.payload.owner)),
+                    partyIdsSet
+                      .contains(PartyId.tryFromProtoPrimitive(locked.payload.amulet.owner)),
+                  amulet =>
+                    partyIdsSet.contains(PartyId.tryFromProtoPrimitive(amulet.payload.owner)),
                 )
             },
           result.afterToken,
@@ -289,15 +294,15 @@ class AcsSnapshotStore(
   def getHoldingsSummary(
       migrationId: Long,
       recordTime: CantonTimestamp,
-      partyIds: Seq[PartyId],
+      partyIds: NonEmptyVector[PartyId],
       asOfRound: Long,
   )(implicit tc: TraceContext): Future[AcsSnapshotStore.HoldingsSummaryResult] = {
     this
       .getHoldingsState(
         migrationId,
         recordTime,
         None,
-        // assumption: the number of contracts is small enough that it will fit in memory
+        // if the limit is exceeded by the results from the DB, an exception will be thrown
         HardLimit.tryCreate(Limit.MaxPageSize),
         partyIds,
       )

apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/store/db/AcsSnapshotStoreTest.scala

@@ -1,20 +1,28 @@
 package org.lfdecentralizedtrust.splice.store.db
 
+import cats.data.NonEmptyVector
 import com.daml.ledger.javaapi.data.Unit as damlUnit
 import com.daml.ledger.javaapi.data.codegen.ContractId
 import org.lfdecentralizedtrust.splice.environment.DarResources
 import org.lfdecentralizedtrust.splice.environment.ledger.api.TransactionTreeUpdate
 import org.lfdecentralizedtrust.splice.migration.DomainMigrationInfo
 import org.lfdecentralizedtrust.splice.scan.store.AcsSnapshotStore
-import org.lfdecentralizedtrust.splice.store.{PageLimit, StoreErrors, StoreTest, UpdateHistory}
+import org.lfdecentralizedtrust.splice.store.{
+  HardLimit,
+  PageLimit,
+  StoreErrors,
+  StoreTest,
+  UpdateHistory,
+}
 import org.lfdecentralizedtrust.splice.util.{Contract, HoldingsSummary, PackageQualifiedName}
 import com.digitalasset.canton.data.CantonTimestamp
 import com.digitalasset.canton.lifecycle.FutureUnlessShutdown
 import com.digitalasset.canton.resource.DbStorage
-import com.digitalasset.canton.topology.{SynchronizerId, PartyId}
+import com.digitalasset.canton.topology.{PartyId, SynchronizerId}
 import com.digitalasset.canton.tracing.TraceContext
 import com.digitalasset.canton.util.MonadUtil
 import com.digitalasset.canton.{HasActorSystem, HasExecutionContext}
+import io.grpc.StatusRuntimeException
 import org.lfdecentralizedtrust.splice.store.UpdateHistory.BackfillingRequirement
 import org.scalatest.Succeeded
 
@@ -490,14 +498,14 @@ class AcsSnapshotStoreTest
             timestamp1,
             None,
             PageLimit.tryCreate(10),
-            Seq(dsoParty),
+            NonEmptyVector.of(dsoParty),
           )
           resultWanteds <- store.getHoldingsState(
             DefaultMigrationId,
             timestamp1,
             None,
             PageLimit.tryCreate(10),
-            Seq(wantedParty1, wantedParty2),
+            NonEmptyVector.of(wantedParty1, wantedParty2),
           )
         } yield {
           resultDso.createdEventsInPage should be(empty)
@@ -526,14 +534,14 @@ class AcsSnapshotStoreTest
             timestamp1,
             None,
             PageLimit.tryCreate(10),
-            Seq(owner),
+            NonEmptyVector.of(owner),
           )
           resultHolder <- store.getHoldingsState(
             DefaultMigrationId,
             timestamp1,
             None,
             PageLimit.tryCreate(10),
-            Seq(holder),
+            NonEmptyVector.of(holder),
           )
         } yield {
           resultHolder.createdEventsInPage should be(empty)
@@ -543,6 +551,39 @@ class AcsSnapshotStoreTest
         }
       }
 
+      "fail if too many contracts were returned by HardLimit" in {
+        val owner = providerParty(1)
+        val holder = providerParty(2)
+        val amulet1 = lockedAmulet(owner, 10, 1L, 0.5)
+        val amulet2 = lockedAmulet(owner, 10, 2L, 0.5)
+        (for {
+          updateHistory <- mkUpdateHistory()
+          store = mkStore(updateHistory)
+          _ <- ingestCreate(
+            updateHistory,
+            amulet1,
+            timestamp1.minusSeconds(10L),
+            Seq(owner, holder),
+          )
+          _ <- ingestCreate(
+            updateHistory,
+            amulet2,
+            timestamp1.minusSeconds(10L),
+            Seq(owner, holder),
+          )
+          _ <- store.insertNewSnapshot(None, DefaultMigrationId, timestamp1)
+          _result <- store.getHoldingsState(
+            DefaultMigrationId,
+            timestamp1,
+            None,
+            HardLimit.tryCreate(1),
+            NonEmptyVector.of(owner),
+          )
+        } yield fail("should not get here, call should've failed")).failed.futureValue shouldBe a[
+          StatusRuntimeException
+        ]
+      }
+
     }
 
     "getHoldingsSummary" should {
@@ -581,19 +622,19 @@ class AcsSnapshotStoreTest
           summaryAtRound3 <- store.getHoldingsSummary(
             DefaultMigrationId,
             timestamp1,
-            Seq(wantedParty1, wantedParty2),
+            NonEmptyVector.of(wantedParty1, wantedParty2),
             asOfRound = 3L,
           )
           summaryAtRound10 <- store.getHoldingsSummary(
             DefaultMigrationId,
             timestamp1,
-            Seq(wantedParty1, wantedParty2),
+            NonEmptyVector.of(wantedParty1, wantedParty2),
             asOfRound = 10L,
           )
           summaryAtRound100 <- store.getHoldingsSummary(
             DefaultMigrationId,
             timestamp1,
-            Seq(wantedParty1, wantedParty2),
+            NonEmptyVector.of(wantedParty1, wantedParty2),
             asOfRound = 100L,
           )
         } yield {

docs/src/release_notes.rst

@@ -19,6 +19,11 @@ Upcoming
     The `/v2/updates` endpoints no longer return the `offset` field in responses,
     and `events_by_id` are now lexicographically ordered by ID for conveniently viewing JSON results.
 
+- Scan
+
+  - Fix a bug where the ``/v0/holdings/summary`` endpoint would return incomplete results when the requested parties had more than 1000 holdings.
+    Additionally, that endpoint and ``/v0/holdings/state`` will now fail if an empty list of parties is provided.
+
 0.4.1
 -----