add BFT enforcement for bulk storage objects

apps/app/src/main/scala/org/lfdecentralizedtrust/splice/console/ScanAppReference.scala

@@ -908,6 +908,18 @@ abstract class ScanAppReference(
       )
     }
 
+  @Help.Summary(
+    "Get checksums for a list of bulk storage objects (using both staging and committed objects)"
+  )
+  def getBulkObjectChecksums(
+      objectKeys: Seq[String]
+  ): definitions.GetBulkObjectChecksumsResponse =
+    consoleEnvironment.run {
+      httpCommand(
+        HttpScanAppClient.GetBulkObjectChecksums(objectKeys)
+      )
+    }
+
   @Help.Summary("Download a bulk storage object")
   def bulkStorageDownload(objectKey: String, output: OutputStream)(implicit
       ec: ExecutionContext,

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

@@ -71,6 +71,8 @@ class ScanTimeBasedIntegrationTest
               updatesPollingInterval = NonNegativeFiniteDuration.ofSeconds(5),
               staging = Some(s3ConfigMock("staging")),
               committed = Some(s3ConfigMock("committed")),
+              bftCheckEnabled =
+                false, // bft checks don't work with a single scan. The bft functionality is tested in the unit test.
             ),
             publicUrl = Some(Uri("http://foo.bar.com")),
           )

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

@@ -1810,6 +1810,30 @@ paths:
         "501":
           $ref: "../../../../common/src/main/openapi/common-external.yaml#/components/responses/501"
 
+  /v0/history/bulk/checksums:
+    post:
+      tags: [ internal, pre-alpha, scan ]
+      x-jvm-package: scan
+      operationId: "getBulkObjectChecksums"
+      description: |
+        **Under Development, do not use in production yet** Get checksums for bulk history objects. Uses both staging and committed objects.
+        Meant for internal use only, as part of the processing pipeline of bulk history objects.
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              "$ref": "#/components/schemas/GetBulkObjectChecksumsRequest"
+      responses:
+        "200":
+          description: ok
+          content:
+            application/json:
+              schema:
+                "$ref": "#/components/schemas/GetBulkObjectChecksumsResponse"
+        "501":
+          $ref: "../../../../common/src/main/openapi/common-external.yaml#/components/responses/501"
+
 
 components:
   schemas:
@@ -4214,6 +4238,33 @@ components:
             to the next `ListBulkUpdateHistoryObjectsRequest` invocation.
             Will be absent when there are no more pages.
 
+    GetBulkObjectChecksumsRequest:
+      type: object
+      required:
+        - object_keys
+      properties:
+        object_keys:
+          description: |
+            The list of keys of the bulk storage objects for which checksums are requested.
+          type: array
+          items:
+            type: string
+
+    GetBulkObjectChecksumsResponse:
+      type: object
+      required:
+        - checksums
+      properties:
+        checksums:
+          description: |
+            The list of checksums for the requested bulk storage objects (in the same order as the object_keys).
+          type: array
+          items:
+            type: object
+            properties:
+              value:
+                type: string
+
     BulkStorageObjectRef:
       type: object
       required:

apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/ScanApp.scala

@@ -269,6 +269,10 @@ class ScanApp(
           retryProvider.metricsFactory,
           config.automation,
           backoffClock = new WallClock(retryProvider.timeouts, loggerFactory),
+          store,
+          svName,
+          ledgerClient,
+          amuletAppParameters.upgradesConfig,
           retryProvider,
           loggerFactory,
         )

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

@@ -37,6 +37,7 @@ import org.lfdecentralizedtrust.splice.environment.{
 import org.lfdecentralizedtrust.splice.http.HttpClient
 import org.lfdecentralizedtrust.splice.http.v0.definitions.{
   AnsEntry,
+  GetBulkObjectChecksumsResponse,
   GetDsoInfoResponse,
   GetRewardAccountingActivityTotalsResponse,
   GetRewardAccountingBatchResponse,
@@ -1120,6 +1121,15 @@ class BftScanConnection(
       )
         .transform(tryBatch => Success(tryBatch.toOption))
   }
+
+  override def getBulkObjectChecksums(
+      objectKeys: Seq[String]
+  )(implicit ec: ExecutionContext, tc: TraceContext): Future[GetBulkObjectChecksumsResponse] =
+    bftCall(
+      _.getBulkObjectChecksums(objectKeys),
+      "getBulkObjectChecksums",
+      consensusFailureLogLevel = Level.DEBUG,
+    )
 }
 trait HasUrl {
   def url: Uri

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

@@ -24,10 +24,12 @@ import org.lfdecentralizedtrust.splice.config.UpgradesConfig
 import org.lfdecentralizedtrust.splice.environment.*
 import org.lfdecentralizedtrust.splice.http.HttpClient
 import org.lfdecentralizedtrust.splice.http.v0.definitions.{
+  GetBulkObjectChecksumsResponse,
   GetDsoInfoResponse,
   GetRewardAccountingActivityTotalsResponse,
   GetRewardAccountingBatchResponse,
   GetRewardAccountingRootHashResponse,
+  HoldingsSummaryRequestV1,
   HoldingsSummaryResponse,
   HoldingsSummaryResponseV1,
   LookupTransferCommandStatusResponse,
@@ -55,7 +57,6 @@ import org.lfdecentralizedtrust.splice.codegen.java.splice.dsorules.{
 }
 import org.lfdecentralizedtrust.splice.http.v0.definitions.HoldingsSummaryRequest.RecordTimeMatch
 import org.lfdecentralizedtrust.splice.metrics.ScanConnectionMetrics
-import org.lfdecentralizedtrust.splice.http.v0.definitions.HoldingsSummaryRequestV1
 
 import scala.concurrent.{ExecutionContext, ExecutionContextExecutor, Future}
 import scala.jdk.OptionConverters.*
@@ -360,6 +361,10 @@ trait ScanConnection
       tc: TraceContext,
   ): Future[Option[GetRewardAccountingBatchResponse]]
 
+  def getBulkObjectChecksums(objectKeys: Seq[String])(implicit
+      ec: ExecutionContext,
+      tc: TraceContext,
+  ): Future[GetBulkObjectChecksumsResponse]
 }
 
 object ScanConnection {

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.{
+  GetBulkObjectChecksumsResponse,
   GetRewardAccountingActivityTotalsResponse,
   GetRewardAccountingBatchResponse,
   GetRewardAccountingRootHashResponse,
@@ -96,7 +97,7 @@ import scala.util.{Failure, Success}
   * to query for the DSO party id.
   */
 class SingleScanConnection private[client] (
-    private[client] val config: ScanAppClientConfig,
+    val config: ScanAppClientConfig,
     upgradesConfig: UpgradesConfig,
     protected val clock: Clock,
     retryProvider: RetryProvider,
@@ -1027,6 +1028,14 @@ class SingleScanConnection private[client] (
       config.adminApi.url,
       HttpScanAppClient.GetRewardAccountingBatch(roundNumber, batchHash),
     )
+
+  override def getBulkObjectChecksums(
+      objectKeys: Seq[String]
+  )(implicit ec: ExecutionContext, tc: TraceContext): Future[GetBulkObjectChecksumsResponse] =
+    runHttpCmd(
+      config.adminApi.url,
+      HttpScanAppClient.GetBulkObjectChecksums(objectKeys),
+    )
 }
 
 object SingleScanConnection {

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

@@ -42,6 +42,7 @@ import org.lfdecentralizedtrust.tokenstandard.{
 }
 import org.lfdecentralizedtrust.splice.http.v0.scan.{
   ForceAcsSnapshotNowResponse,
+  GetBulkObjectChecksumsResponse,
   GetDateOfFirstSnapshotAfterResponse,
   GetDateOfMostRecentSnapshotBeforeResponse,
   GetLsuResponse,
@@ -3371,6 +3372,32 @@ object HttpScanAppClient {
     }
   }
 
+  case class GetBulkObjectChecksums(
+      objectKeys: Seq[String]
+  ) extends InternalBaseCommand[
+        http.GetBulkObjectChecksumsResponse,
+        definitions.GetBulkObjectChecksumsResponse,
+      ] {
+    override def submitRequest(
+        client: Client,
+        headers: List[HttpHeader],
+    ): EitherT[Future, Either[Throwable, HttpResponse], GetBulkObjectChecksumsResponse] =
+      client.getBulkObjectChecksums(
+        definitions.GetBulkObjectChecksumsRequest(objectKeys.toVector),
+        headers,
+      )
+
+    override protected def handleOk()(implicit
+        decoder: TemplateJsonDecoder
+    ): PartialFunction[GetBulkObjectChecksumsResponse, Either[
+      String,
+      definitions.GetBulkObjectChecksumsResponse,
+    ]] = {
+      case http.GetBulkObjectChecksumsResponse.OK(response) => Right(response)
+      case http.GetBulkObjectChecksumsResponse.NotImplemented(err) => Left(err.error)
+    }
+  }
+
   case class BulkStorageDownload(
       objectKey: String
   ) extends ScanStreamBaseCommand[

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

@@ -68,10 +68,11 @@ import org.lfdecentralizedtrust.splice.http.v0.{definitions, scan as v0}
 import org.lfdecentralizedtrust.splice.http.v0.definitions.{
   AcsRequest,
   BatchListVotesByVoteRequestsRequest,
-  DamlValueEncoding,
   CountVoteResultsRequest,
+  DamlValueEncoding,
   ErrorResponse,
   EventHistoryRequest,
+  GetBulkObjectChecksumsRequest,
   HoldingsStateRequest,
   HoldingsSummaryRequest,
   HoldingsSummaryRequestV1,
@@ -2649,6 +2650,29 @@ class HttpScanHandler(
     }
   }
 
+  override def getBulkObjectChecksums(respond: ScanResource.GetBulkObjectChecksumsResponse.type)(
+      body: GetBulkObjectChecksumsRequest
+  )(extracted: TraceContext): Future[ScanResource.GetBulkObjectChecksumsResponse] = {
+    implicit val tc = extracted
+    withSpan(s"$workflowId.getBulkObjectChecksums") { _ => _ =>
+      bulkStorage.fold(
+        Future.failed[ScanResource.GetBulkObjectChecksumsResponse](
+          Status.UNIMPLEMENTED
+            .withDescription("Bulk storage is not configured")
+            .asRuntimeException()
+        )
+      ) { bulkStorage =>
+        bulkStorage.getObjectChecksums(body.objectKeys).map { checksums =>
+          ScanResource.GetBulkObjectChecksumsResponse.OK(
+            definitions.GetBulkObjectChecksumsResponse(
+              checksums.map(definitions.GetBulkObjectChecksumsResponse.Checksums(_)).toVector
+            )
+          )
+        }
+      }
+    }
+  }
+
   def getRollForwardLsu(respond: ScanResource.GetRollForwardLsuResponse.type)()(
       extracted: TraceContext
   ): Future[ScanResource.GetRollForwardLsuResponse] = {

apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/automation/ScanHistoryBackfillingTrigger.scala

@@ -18,7 +18,6 @@ import org.lfdecentralizedtrust.splice.scan.admin.api.client.{
   BackfillingScanConnection,
   BftScanConnection,
 }
-import org.lfdecentralizedtrust.splice.scan.config.ScanAppClientConfig
 import org.lfdecentralizedtrust.splice.scan.store.ScanHistoryBackfilling.{
   FoundingTransactionTreeUpdate,
   InitialTransactionTreeUpdate,
@@ -35,12 +34,13 @@ import org.lfdecentralizedtrust.splice.store.{
 }
 import org.lfdecentralizedtrust.splice.util.TemplateJsonDecoder
 import com.digitalasset.canton.data.CantonTimestamp
-import com.digitalasset.canton.lifecycle.{AsyncOrSyncCloseable, SyncCloseable}
+import com.digitalasset.canton.lifecycle.AsyncOrSyncCloseable
 import com.digitalasset.canton.logging.pretty.{Pretty, PrettyPrinting}
 import com.digitalasset.canton.topology.PartyId
 import com.digitalasset.canton.tracing.TraceContext
 import io.opentelemetry.api.trace.Tracer
 import org.apache.pekko.stream.Materializer
+import org.lfdecentralizedtrust.splice.scan.util.HasPeerBftScanConnection
 import org.lfdecentralizedtrust.splice.store.UpdateHistory.BackfillingState
 
 import scala.concurrent.{ExecutionContextExecutor, Future, blocking}
@@ -61,7 +61,8 @@ class ScanHistoryBackfillingTrigger(
     httpClient: HttpClient,
     templateJsonDecoder: TemplateJsonDecoder,
     mat: Materializer,
-) extends PollingParallelTaskExecutionTrigger[ScanHistoryBackfillingTrigger.Task] {
+) extends PollingParallelTaskExecutionTrigger[ScanHistoryBackfillingTrigger.Task]
+    with HasPeerBftScanConnection {
 
   private val currentMigrationId = updateHistory.domainMigrationId
 
@@ -79,10 +80,6 @@ class ScanHistoryBackfillingTrigger(
   @volatile
   private var findHistoryStartAfter: Option[(Long, CantonTimestamp)] = None
 
-  @SuppressWarnings(Array("org.wartremover.warts.Var"))
-  @volatile
-  private var connectionVar: Option[BftScanConnection] = None
-
   @SuppressWarnings(Array("org.wartremover.warts.Var"))
   @volatile
   private var backfillingVar: Option[ScanHistoryBackfilling] = None
@@ -222,34 +219,6 @@ class ScanHistoryBackfillingTrigger(
     }
   }
 
-  private def getOrCreateScanConnection()(implicit tc: TraceContext): Future[BftScanConnection] =
-    blocking {
-      mutex.exclusive {
-        connectionVar match {
-          case Some(connection) =>
-            Future.successful(connection)
-          case None =>
-            for {
-              connection <- BftScanConnection.peerScanConnection(
-                () => BftScanConnection.Bft.getPeerScansFromStore(store, svName),
-                ledgerClient,
-                // When the network is starting up, the pool of SVs is changing fast
-                // Using a short refresh interval to quickly pick up new SVs
-                scansRefreshInterval = context.config.pollingInterval,
-                amuletRulesCacheTimeToLive = ScanAppClientConfig.DefaultAmuletRulesCacheTimeToLive,
-                upgradesConfig,
-                context.clock,
-                context.retryProvider,
-                loggerFactory,
-              )
-            } yield {
-              connectionVar = Some(connection)
-              connection
-            }
-        }
-      }
-    }
-
   private def getOrCreateBackfilling(
       connection: BackfillingScanConnection
   ): ScanHistoryBackfilling = blocking {
@@ -272,6 +241,23 @@ class ScanHistoryBackfillingTrigger(
     }
   }
 
+  private def getOrCreateScanConnection()(implicit
+      tc: TraceContext,
+      ec: ExecutionContextExecutor,
+      mat: Materializer,
+      httpClient: HttpClient,
+      templateJsonDecoder: TemplateJsonDecoder,
+  ): Future[BftScanConnection] = getOrCreateScanConnection(
+    store,
+    svName,
+    ledgerClient,
+    context.config,
+    upgradesConfig,
+    context.clock,
+    context.retryProvider,
+    loggerFactory,
+  )
+
   private def performBackfilling()(implicit traceContext: TraceContext): Future[TaskOutcome] = for {
     connection <- getOrCreateScanConnection()
     backfilling = getOrCreateBackfilling(connection)
@@ -327,14 +313,7 @@ class ScanHistoryBackfillingTrigger(
   } yield outcome
 
   override def closeAsync(): Seq[AsyncOrSyncCloseable] = {
-    connectionVar
-      .map(connection =>
-        SyncCloseable(
-          "closing scan connection",
-          connection.close(),
-        )
-      )
-      .toList
+    closeScanConnection().toList
   }
 }