diff --git a/apps/app/src/main/scala/org/lfdecentralizedtrust/splice/console/ScanAppReference.scala b/apps/app/src/main/scala/org/lfdecentralizedtrust/splice/console/ScanAppReference.scala
index e510e5b33a..99226c2c5a 100644
--- a/apps/app/src/main/scala/org/lfdecentralizedtrust/splice/console/ScanAppReference.scala
+++ b/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,
diff --git a/apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/tests/ScanTimeBasedIntegrationTest.scala b/apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/tests/ScanTimeBasedIntegrationTest.scala
index ea4371ab30..1b059b2df8 100644
--- a/apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/tests/ScanTimeBasedIntegrationTest.scala
+++ b/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")),
           )
diff --git a/apps/scan/src/main/openapi/scan.yaml b/apps/scan/src/main/openapi/scan.yaml
index 527fe5fceb..78c982d486 100644
--- a/apps/scan/src/main/openapi/scan.yaml
+++ b/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:
diff --git a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/ScanApp.scala b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/ScanApp.scala
index 5cfe864f60..db692e97e0 100644
--- a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/ScanApp.scala
+++ b/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,
         )
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..7addee63da 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
@@ -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
diff --git a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/admin/api/client/ScanConnection.scala b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/admin/api/client/ScanConnection.scala
index caed44de87..bbf758cb1c 100644
--- a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/admin/api/client/ScanConnection.scala
+++ b/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 {
diff --git a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/admin/api/client/SingleScanConnection.scala b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/admin/api/client/SingleScanConnection.scala
index f9d818c9d7..e0852b971a 100644
--- a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/admin/api/client/SingleScanConnection.scala
+++ b/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 {
diff --git a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/admin/api/client/commands/HttpScanAppClient.scala b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/admin/api/client/commands/HttpScanAppClient.scala
index 28a6bf83d0..612282b71f 100644
--- a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/admin/api/client/commands/HttpScanAppClient.scala
+++ b/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[
diff --git a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/admin/http/HttpScanHandler.scala b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/admin/http/HttpScanHandler.scala
index b5e2302bf0..88fd4fd880 100644
--- a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/admin/http/HttpScanHandler.scala
+++ b/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] = {
diff --git a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/automation/ScanHistoryBackfillingTrigger.scala b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/automation/ScanHistoryBackfillingTrigger.scala
index 3876147017..0944dd72e8 100644
--- a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/automation/ScanHistoryBackfillingTrigger.scala
+++ b/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
   }
 }
 
diff --git a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/config/ScanAppConfig.scala b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/config/ScanAppConfig.scala
index d9c87180c5..9825eb02e3 100644
--- a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/config/ScanAppConfig.scala
+++ b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/config/ScanAppConfig.scala
@@ -47,6 +47,7 @@ final case class BulkStorageConfig(
     maxParallelPartUploads: Int = 4,
     staging: Option[S3Config] = None,
     committed: Option[S3Config] = None,
+    bftCheckEnabled: Boolean = true,
 )
 
 /** @param miningRoundsCacheTimeToLiveOverride Intended only for testing!
diff --git a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/bulk/AcsSnapshotBulkStorageCommitFromStaging.scala b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/bulk/AcsSnapshotBulkStorageCommitFromStaging.scala
index fba378264d..206edbbd14 100644
--- a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/bulk/AcsSnapshotBulkStorageCommitFromStaging.scala
+++ b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/bulk/AcsSnapshotBulkStorageCommitFromStaging.scala
@@ -4,24 +4,40 @@
 package org.lfdecentralizedtrust.splice.scan.store.bulk
 
 import com.digitalasset.canton.logging.{NamedLoggerFactory, NamedLogging}
+import com.digitalasset.canton.time.Clock
 import com.digitalasset.canton.tracing.TraceContext
 import io.grpc.{Status, StatusRuntimeException}
 import org.apache.pekko.NotUsed
 import org.apache.pekko.actor.ActorSystem
 import org.apache.pekko.stream.scaladsl.Flow
+import org.lfdecentralizedtrust.splice.config.{AutomationConfig, UpgradesConfig}
+import org.lfdecentralizedtrust.splice.environment.{RetryProvider, SpliceLedgerClient}
+import org.lfdecentralizedtrust.splice.http.HttpClient
 import org.lfdecentralizedtrust.splice.scan.config.BulkStorageConfig
+import org.lfdecentralizedtrust.splice.scan.store.ScanStore
 import org.lfdecentralizedtrust.splice.store.{S3BucketConnection, TimestampWithMigrationId}
+import org.lfdecentralizedtrust.splice.util.TemplateJsonDecoder
 
-import scala.concurrent.{ExecutionContext, Future}
+import scala.concurrent.{ExecutionContextExecutor, Future}
 
 class AcsSnapshotBulkStorageCommitFromStaging(
     stagingS3Connection: S3BucketConnection,
     committedS3Connection: S3BucketConnection,
     bulkStorageReader: BulkStorageReader,
     appConfig: BulkStorageConfig,
+    store: ScanStore,
+    svName: String,
+    ledgerClient: SpliceLedgerClient,
+    automationConfig: AutomationConfig,
+    upgradesConfig: UpgradesConfig,
+    clock: Clock,
+    retryProvider: RetryProvider,
     val loggerFactory: NamedLoggerFactory,
-)(implicit ec: ExecutionContext)
-    extends AcsSnapshotBulkStorageWriter
+)(implicit
+    ec: ExecutionContextExecutor,
+    httpClient: HttpClient,
+    templateJsonDecoder: TemplateJsonDecoder,
+) extends AcsSnapshotBulkStorageWriter
     with NamedLogging {
 
   override def getNextSnapshotTimestampAfter(
@@ -55,6 +71,13 @@ class AcsSnapshotBulkStorageCommitFromStaging(
               Seq.empty
           },
       appConfig,
+      store,
+      svName,
+      ledgerClient,
+      automationConfig,
+      upgradesConfig,
+      clock,
+      retryProvider,
       loggerFactory,
     )
   }
diff --git a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/bulk/BulkStorage.scala b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/bulk/BulkStorage.scala
index 1ed4ff4b55..5e57112a0f 100644
--- a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/bulk/BulkStorage.scala
+++ b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/bulk/BulkStorage.scala
@@ -11,16 +11,21 @@ import com.digitalasset.canton.tracing.TraceContext
 import io.grpc.Status
 import io.opentelemetry.api.trace.Tracer
 import org.apache.pekko.actor.{ActorSystem, Cancellable}
-import org.lfdecentralizedtrust.splice.config.{AutomationConfig, S3Config}
-import org.lfdecentralizedtrust.splice.environment.RetryProvider
+import org.lfdecentralizedtrust.splice.config.{AutomationConfig, S3Config, UpgradesConfig}
+import org.lfdecentralizedtrust.splice.environment.{RetryProvider, SpliceLedgerClient}
 import org.lfdecentralizedtrust.splice.scan.config.{BulkStorageConfig, ScanStorageConfig}
-import org.lfdecentralizedtrust.splice.scan.store.{AcsSnapshotStore, ScanKeyValueProvider}
+import org.lfdecentralizedtrust.splice.scan.store.{
+  AcsSnapshotStore,
+  ScanKeyValueProvider,
+  ScanStore,
+}
 import org.lfdecentralizedtrust.splice.store.{HistoryMetrics, S3BucketConnection, UpdateHistory}
 
-import scala.concurrent.{ExecutionContext, Future}
+import scala.concurrent.{ExecutionContextExecutor, Future}
 import cats.implicits.*
 import org.apache.pekko.stream.scaladsl.Source
 import org.lfdecentralizedtrust.splice.PekkoRetryableService
+import org.lfdecentralizedtrust.splice.http.HttpClient
 import org.lfdecentralizedtrust.splice.scan.store.bulk.BulkStorage.{
   acsCommittedKvStoreKey,
   acsStagingKvStoreKey,
@@ -28,6 +33,7 @@ import org.lfdecentralizedtrust.splice.scan.store.bulk.BulkStorage.{
   updatesCommittedKvStoreKey,
   updatesStagingKvStoreKey,
 }
+import org.lfdecentralizedtrust.splice.util.TemplateJsonDecoder
 
 import scala.concurrent.duration.*
 
@@ -43,13 +49,19 @@ class BulkStorage(
     metricsFactory: LabeledMetricsFactory,
     automationConfig: AutomationConfig,
     backoffClock: Clock,
+    store: ScanStore,
+    svName: String,
+    ledgerClient: SpliceLedgerClient,
+    upgradesConfig: UpgradesConfig,
     override val retryProvider: RetryProvider,
     override val loggerFactory: NamedLoggerFactory,
 )(implicit
     actorSystem: ActorSystem,
     tc: TraceContext,
-    ec: ExecutionContext,
+    ec: ExecutionContextExecutor,
     tracer: Tracer,
+    httpClient: HttpClient,
+    templateJsonDecoder: TemplateJsonDecoder,
 ) extends NamedLogging
     with FlagCloseableAsync
     with RetryProvider.Has {
@@ -128,6 +140,13 @@ class BulkStorage(
     committedConnection,
     reader,
     appConfig,
+    store,
+    svName,
+    ledgerClient,
+    automationConfig,
+    upgradesConfig,
+    backoffClock,
+    retryProvider,
     loggerFactory,
   )
   val acsCommitted = new AcsSnapshotBulkStorage(
@@ -160,6 +179,13 @@ class BulkStorage(
     committedConnection,
     reader,
     appConfig,
+    store,
+    svName,
+    ledgerClient,
+    automationConfig,
+    upgradesConfig,
+    backoffClock,
+    retryProvider,
     loggerFactory,
   )
   val updatesCommitted = new UpdateHistoryBulkStorage(
@@ -197,13 +223,19 @@ object BulkStorage {
       metricsFactory: LabeledMetricsFactory,
       automationConfig: AutomationConfig,
       backoffClock: Clock,
+      store: ScanStore,
+      svName: String,
+      ledgerClient: SpliceLedgerClient,
+      upgradesConfig: UpgradesConfig,
       retryProvider: RetryProvider,
       loggerFactory: NamedLoggerFactory,
   )(implicit
       actorSystem: ActorSystem,
       tc: TraceContext,
-      ec: ExecutionContext,
+      ec: ExecutionContextExecutor,
       tracer: Tracer,
+      httpClient: HttpClient,
+      templateJsonDecoder: TemplateJsonDecoder,
   ): BulkStorage = {
     val logger = loggerFactory.getTracedLogger(classOf[BulkStorage])
 
@@ -225,6 +257,10 @@ object BulkStorage {
         metricsFactory,
         automationConfig,
         backoffClock,
+        store,
+        svName,
+        ledgerClient,
+        upgradesConfig,
         retryProvider,
         loggerFactory,
       )
diff --git a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/bulk/BulkStorageCommitFromStaging.scala b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/bulk/BulkStorageCommitFromStaging.scala
index 75f18af458..bf2a99b1b5 100644
--- a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/bulk/BulkStorageCommitFromStaging.scala
+++ b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/bulk/BulkStorageCommitFromStaging.scala
@@ -4,15 +4,24 @@
 package org.lfdecentralizedtrust.splice.scan.store.bulk
 
 import com.digitalasset.canton.logging.{NamedLoggerFactory, NamedLogging}
+import com.digitalasset.canton.time.Clock
 import com.digitalasset.canton.tracing.TraceContext
 import org.apache.pekko.NotUsed
 import org.apache.pekko.actor.ActorSystem
-import org.apache.pekko.stream.scaladsl.{Flow, Sink, Source}
+import org.apache.pekko.http.scaladsl.model.StatusCodes
+import org.apache.pekko.stream.scaladsl.{Flow, Source}
+import org.lfdecentralizedtrust.splice.admin.http.HttpErrorWithHttpCode
+import org.lfdecentralizedtrust.splice.config.{AutomationConfig, UpgradesConfig}
+import org.lfdecentralizedtrust.splice.environment.{RetryProvider, SpliceLedgerClient}
+import org.lfdecentralizedtrust.splice.http.HttpClient
 import org.lfdecentralizedtrust.splice.scan.config.BulkStorageConfig
+import org.lfdecentralizedtrust.splice.scan.store.ScanStore
+import org.lfdecentralizedtrust.splice.scan.util.HasPeerBftScanConnection
 import org.lfdecentralizedtrust.splice.store.S3BucketConnection
 import org.lfdecentralizedtrust.splice.store.S3BucketConnection.ObjectKeyAndChecksum
+import org.lfdecentralizedtrust.splice.util.TemplateJsonDecoder
 
-import scala.concurrent.{ExecutionContext, Future}
+import scala.concurrent.{ExecutionContextExecutor, Future}
 
 // TODO(#5884): review parallelism here. We use parallelism = 1 all over, but unsure whether that's actually necessary.
 
@@ -21,19 +30,75 @@ class BulkStorageCommitFromStaging[T](
     committedS3Connection: S3BucketConnection,
     getObjects: T => Future[Seq[ObjectKeyAndChecksum]],
     appConfig: BulkStorageConfig,
+    store: ScanStore,
+    svName: String,
+    ledgerClient: SpliceLedgerClient,
+    automationConfig: AutomationConfig,
+    upgradesConfig: UpgradesConfig,
+    clock: Clock,
+    retryProvider: RetryProvider,
     override val loggerFactory: NamedLoggerFactory,
 )(implicit
     tc: TraceContext,
-    ec: ExecutionContext,
+    ec: ExecutionContextExecutor,
     actorSystem: ActorSystem,
-) extends NamedLogging {
+    httpClient: HttpClient,
+    templateJsonDecoder: TemplateJsonDecoder,
+) extends HasPeerBftScanConnection
+    with NamedLogging {
   private def checkBftForObjects(
       objects: Seq[ObjectKeyAndChecksum]
   ): Future[Boolean] = {
     logger.debug(
       s"Checking BFT agreement for objects: ${objects.map(_.key).mkString(", ")}"
     )
-    Future.successful(true)
+    if (appConfig.bftCheckEnabled) {
+      for {
+        connection <- getOrCreateScanConnection(
+          store,
+          svName,
+          ledgerClient,
+          automationConfig,
+          upgradesConfig,
+          clock: Clock,
+          retryProvider,
+          loggerFactory,
+        )
+        bft <- connection.getBulkObjectChecksums(objects.map(_.key)).map(Some(_)).recoverWith {
+          case ex @ HttpErrorWithHttpCode(code, _) =>
+            if (code == StatusCodes.BadGateway) {
+              Future.successful(None)
+            } else {
+              throw ex
+            }
+        }
+      } yield {
+        bft match {
+          case Some(bftChecksums) =>
+            val consensusChecksums = bftChecksums.checksums.filter(_.value.isDefined)
+            logger.debug(
+              s"Consensus achieved on ${consensusChecksums.length} out of ${objects.length} objects"
+            )
+            val consensus =
+              bftChecksums.checksums.filter(_.value.isDefined).map(_.value) == objects.map(oc =>
+                Some(oc.checksum)
+              )
+            if (consensusChecksums.length == objects.length && !consensus) {
+              logger.error(
+                s"All objects are known to the BFT peers, but the checksums do not match. This indicates an error in the actual data generated for bulk storage. Expected: ${objects
+                    .map(_.checksum)
+                    .mkString(", ")}, got: ${consensusChecksums.mkString(", ")}"
+              )
+            }
+            consensus
+          case None =>
+            false
+        }
+      }
+    } else {
+      logger.trace("BFT check is disabled, skipping BFT agreement check")
+      Future.successful(true)
+    }
   }
   // TODO(#5884): implement the BFT check
 
@@ -42,7 +107,7 @@ class BulkStorageCommitFromStaging[T](
     (T, Seq[ObjectKeyAndChecksum]),
     NotUsed,
   ] = {
-    Flow[(T, Seq[ObjectKeyAndChecksum])].mapAsync(parallelism = 1) { case (t, obj) =>
+    Flow[(T, Seq[ObjectKeyAndChecksum])].flatMapConcat { case (t, obj) =>
       Source
         .repeat(obj)
         .mapAsync(parallelism = 1)(obj => checkBftForObjects(obj).map(result => (obj, result)))
@@ -60,8 +125,7 @@ class BulkStorageCommitFromStaging[T](
             Source.single((obj, false)).delay(appConfig.bftRetryInterval.underlying)
         }
         .takeWhile({ case (_, bftReached) => !bftReached }, inclusive = true)
-        .runWith(Sink.last)
-        .map { case (obj, _) => (t, obj) }
+        .collect { case (o, true) => (t, o) }
     }
   }
 
@@ -136,17 +200,33 @@ object BulkStorageCommitFromStaging {
       committedS3Connection: S3BucketConnection,
       getStagingObjects: T => Future[Seq[ObjectKeyAndChecksum]],
       appConfig: BulkStorageConfig,
+      store: ScanStore,
+      svName: String,
+      ledgerClient: SpliceLedgerClient,
+      automationConfig: AutomationConfig,
+      upgradesConfig: UpgradesConfig,
+      clock: Clock,
+      retryProvider: RetryProvider,
       loggerFactory: NamedLoggerFactory,
   )(implicit
       tc: TraceContext,
-      ec: ExecutionContext,
+      ec: ExecutionContextExecutor,
       actorSystem: ActorSystem,
+      httpClient: HttpClient,
+      templateJsonDecoder: TemplateJsonDecoder,
   ): Flow[T, T, NotUsed] = {
     new BulkStorageCommitFromStaging[T](
       stagingS3Connection,
       committedS3Connection,
       getStagingObjects,
       appConfig,
+      store,
+      svName,
+      ledgerClient,
+      automationConfig,
+      upgradesConfig,
+      clock,
+      retryProvider,
       loggerFactory,
     ).getFlow
   }
diff --git a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/bulk/BulkStorageReader.scala b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/bulk/BulkStorageReader.scala
index de1653fb35..234f8969df 100644
--- a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/bulk/BulkStorageReader.scala
+++ b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/bulk/BulkStorageReader.scala
@@ -143,6 +143,19 @@ class BulkStorageReader(
       }
   }
 
+  def getObjectChecksums(
+      objectKeys: Seq[String]
+  ): Future[Seq[Option[String]]] = {
+    for {
+      committed <- committedS3Connection.getChecksums(objectKeys)
+      staging <- stagingS3Connection.getChecksums(objectKeys)
+    } yield {
+      objectKeys.map { key =>
+        committed.find(_.key == key).orElse(staging.find(_.key == key)).map(_.checksum)
+      }
+    }
+  }
+
   private def getSegmentStartingAt(
       startTimestamp: Option[CantonTimestamp]
   ): Future[Option[(CantonTimestamp, CantonTimestamp)]] =
diff --git a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/bulk/UpdateHistoryBulkStorageCommitFromStaging.scala b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/bulk/UpdateHistoryBulkStorageCommitFromStaging.scala
index 1911f9f893..e5452cb798 100644
--- a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/bulk/UpdateHistoryBulkStorageCommitFromStaging.scala
+++ b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/bulk/UpdateHistoryBulkStorageCommitFromStaging.scala
@@ -4,24 +4,41 @@
 package org.lfdecentralizedtrust.splice.scan.store.bulk
 
 import com.digitalasset.canton.logging.{NamedLoggerFactory, NamedLogging}
+import com.digitalasset.canton.time.Clock
 import com.digitalasset.canton.tracing.TraceContext
 import io.grpc.{Status, StatusRuntimeException}
 import org.apache.pekko.NotUsed
 import org.apache.pekko.actor.ActorSystem
 import org.apache.pekko.stream.scaladsl.Flow
+import org.lfdecentralizedtrust.splice.config.{AutomationConfig, UpgradesConfig}
+import org.lfdecentralizedtrust.splice.environment.{RetryProvider, SpliceLedgerClient}
+import org.lfdecentralizedtrust.splice.http.HttpClient
 import org.lfdecentralizedtrust.splice.scan.config.BulkStorageConfig
+import org.lfdecentralizedtrust.splice.scan.store.ScanStore
 import org.lfdecentralizedtrust.splice.store.{S3BucketConnection, TimestampWithMigrationId}
+import org.lfdecentralizedtrust.splice.util.TemplateJsonDecoder
 
-import scala.concurrent.{ExecutionContext, Future}
+import scala.concurrent.{ExecutionContextExecutor, Future}
 
 class UpdateHistoryBulkStorageCommitFromStaging(
     stagingS3Connection: S3BucketConnection,
     committedS3Connection: S3BucketConnection,
     bulkStorageReader: BulkStorageReader,
     appConfig: BulkStorageConfig,
+    store: ScanStore,
+    svName: String,
+    ledgerClient: SpliceLedgerClient,
+    automationConfig: AutomationConfig,
+    upgradesConfig: UpgradesConfig,
+    clock: Clock,
+    retryProvider: RetryProvider,
     val loggerFactory: NamedLoggerFactory,
-)(implicit ec: ExecutionContext, actorSystem: ActorSystem)
-    extends UpdateHistoryBulkStorageWriter
+)(implicit
+    ec: ExecutionContextExecutor,
+    actorSystem: ActorSystem,
+    httpClient: HttpClient,
+    templateJsonDecoder: TemplateJsonDecoder,
+) extends UpdateHistoryBulkStorageWriter
     with NamedLogging {
   override def processSegmentsFlow(implicit
       tc: TraceContext
@@ -39,6 +56,13 @@ class UpdateHistoryBulkStorageCommitFromStaging(
               Seq.empty
           },
       appConfig,
+      store,
+      svName,
+      ledgerClient,
+      automationConfig,
+      upgradesConfig,
+      clock,
+      retryProvider,
       loggerFactory,
     )
 
diff --git a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/util/HasPeerBftScanConnection.scala b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/util/HasPeerBftScanConnection.scala
new file mode 100644
index 0000000000..bed8bcfc0e
--- /dev/null
+++ b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/util/HasPeerBftScanConnection.scala
@@ -0,0 +1,82 @@
+// Copyright (c) 2024 Digital Asset (Switzerland) GmbH and/or its affiliates. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package org.lfdecentralizedtrust.splice.scan.util
+
+import com.digitalasset.canton.lifecycle.SyncCloseable
+import com.digitalasset.canton.logging.NamedLoggerFactory
+import com.digitalasset.canton.time.Clock
+import com.digitalasset.canton.tracing.TraceContext
+import com.digitalasset.canton.util.Mutex
+import org.apache.pekko.stream.Materializer
+import org.lfdecentralizedtrust.splice.config.{AutomationConfig, UpgradesConfig}
+import org.lfdecentralizedtrust.splice.environment.{RetryProvider, SpliceLedgerClient}
+import org.lfdecentralizedtrust.splice.http.HttpClient
+import org.lfdecentralizedtrust.splice.scan.admin.api.client.BftScanConnection
+import org.lfdecentralizedtrust.splice.scan.config.ScanAppClientConfig
+import org.lfdecentralizedtrust.splice.scan.store.ScanStore
+import org.lfdecentralizedtrust.splice.util.TemplateJsonDecoder
+
+import scala.concurrent.{ExecutionContextExecutor, Future, blocking}
+
+trait HasPeerBftScanConnection {
+
+  private val mutex = Mutex()
+
+  @SuppressWarnings(Array("org.wartremover.warts.Var"))
+  @volatile
+  private var connectionVar: Option[BftScanConnection] = None
+
+  protected def getOrCreateScanConnection(
+      store: ScanStore,
+      svName: String,
+      ledgerClient: SpliceLedgerClient,
+      automationConfig: AutomationConfig,
+      upgradesConfig: UpgradesConfig,
+      clock: Clock,
+      retryProvider: RetryProvider,
+      loggerFactory: NamedLoggerFactory,
+  )(implicit
+      tc: TraceContext,
+      ec: ExecutionContextExecutor,
+      mat: Materializer,
+      httpClient: HttpClient,
+      templateJsonDecoder: TemplateJsonDecoder,
+  ): 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 = automationConfig.pollingInterval,
+                amuletRulesCacheTimeToLive = ScanAppClientConfig.DefaultAmuletRulesCacheTimeToLive,
+                upgradesConfig,
+                clock,
+                retryProvider,
+                loggerFactory,
+              )
+            } yield {
+              connectionVar = Some(connection)
+              connection
+            }
+        }
+      }
+    }
+
+  protected def closeScanConnection(): Option[SyncCloseable] =
+    connectionVar
+      .map(connection =>
+        SyncCloseable(
+          "closing scan connection",
+          connection.close(),
+        )
+      )
+
+}
diff --git a/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/store/bulk/AcsSnapshotBulkStorageCommitFromStagingTest.scala b/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/store/bulk/AcsSnapshotBulkStorageCommitFromStagingTest.scala
index 44adcd701e..af948c5137 100644
--- a/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/store/bulk/AcsSnapshotBulkStorageCommitFromStagingTest.scala
+++ b/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/store/bulk/AcsSnapshotBulkStorageCommitFromStagingTest.scala
@@ -19,6 +19,7 @@ import org.apache.pekko.actor.Cancellable
 import org.apache.pekko.stream.scaladsl.Source
 import org.lfdecentralizedtrust.splice.config.AutomationConfig
 import org.lfdecentralizedtrust.splice.environment.RetryProvider
+import org.lfdecentralizedtrust.splice.http.HttpClient
 import org.lfdecentralizedtrust.splice.scan.config.{BulkStorageConfig, ScanStorageConfig}
 import org.lfdecentralizedtrust.splice.scan.store.{ScanKeyValueProvider, ScanKeyValueStore}
 import org.lfdecentralizedtrust.splice.store.{
@@ -31,6 +32,7 @@ import org.lfdecentralizedtrust.splice.store.{
 import scala.concurrent.Future
 import scala.util.Using
 import org.lfdecentralizedtrust.splice.store.db.SplicePostgresTest
+import org.lfdecentralizedtrust.splice.util.TemplateJsonDecoder
 import org.scalatest.Assertion
 import org.slf4j.event.Level
 
@@ -52,7 +54,8 @@ class AcsSnapshotBulkStorageCommitFromStagingTest
     zstdCompressionLevel = 3,
   )
   val appConfig = BulkStorageConfig(
-    snapshotPollingInterval = NonNegativeFiniteDuration.ofSeconds(5)
+    snapshotPollingInterval = NonNegativeFiniteDuration.ofSeconds(5),
+    bftCheckEnabled = false, // bft checks are tested elsewhere
   )
 
   override val initialBuckets: Seq[String] = Seq("staging", "committed")
@@ -97,6 +100,9 @@ class AcsSnapshotBulkStorageCommitFromStagingTest
       )
 
       def withNewCommitService(body: => Assertion): Assertion = {
+        implicit val httpClient: HttpClient = null // not used when bft reads are disabled
+        implicit val templateJsonDecoder: TemplateJsonDecoder =
+          null // not used when bft reads are disabled
         val retryProvider =
           RetryProvider(loggerFactory, timeouts, FutureSupervisor.Noop, NoOpMetricsFactory)
         val acsCommittedWriter = new AcsSnapshotBulkStorageCommitFromStaging(
@@ -104,6 +110,13 @@ class AcsSnapshotBulkStorageCommitFromStagingTest
           committedConnection,
           reader,
           appConfig,
+          null, // not used when bft reads are disabled
+          null, // not used when bft reads are disabled
+          null, // not used when bft reads are disabled
+          null, // not used when bft reads are disabled
+          null, // not used when bft reads are disabled
+          null, // not used when bft reads are disabled
+          null, // not used when bft reads are disabled
           loggerFactory,
         )
         val commitService = {
diff --git a/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/store/bulk/BulkStorageCommitFromStagingTest.scala b/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/store/bulk/BulkStorageCommitFromStagingTest.scala
index be368bc401..12fb788124 100644
--- a/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/store/bulk/BulkStorageCommitFromStagingTest.scala
+++ b/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/store/bulk/BulkStorageCommitFromStagingTest.scala
@@ -3,40 +3,68 @@
 
 package org.lfdecentralizedtrust.splice.scan.store.bulk
 
+import com.digitalasset.canton.config.NonNegativeFiniteDuration
 import com.digitalasset.canton.lifecycle.FutureUnlessShutdown
-import com.digitalasset.canton.logging.SuppressionRule
+import com.digitalasset.canton.logging.{NamedLoggerFactory, SuppressionRule}
 import com.digitalasset.canton.resource.DbStorage
+import com.digitalasset.canton.time.Clock
+import com.digitalasset.canton.topology.SynchronizerId
 import com.digitalasset.canton.tracing.TraceContext
 import com.digitalasset.canton.{HasActorSystem, HasExecutionContext}
+import org.apache.pekko.NotUsed
+import org.apache.pekko.http.scaladsl.model.Uri
+import org.lfdecentralizedtrust.splice.config.NetworkAppClientConfig
+import org.lfdecentralizedtrust.splice.scan.config.ScanAppClientConfig
+import org.lfdecentralizedtrust.splice.test.HasRetryProvider
+import org.apache.pekko.stream.Materializer
 import org.slf4j.event.Level
-import org.apache.pekko.stream.scaladsl.Keep
+import org.apache.pekko.stream.scaladsl.{Flow, Keep}
 import org.apache.pekko.stream.testkit.scaladsl.{TestSink, TestSource}
+import org.lfdecentralizedtrust.splice.config.{AutomationConfig, UpgradesConfig}
+import org.lfdecentralizedtrust.splice.environment.{RetryProvider, SpliceLedgerClient}
+import org.lfdecentralizedtrust.splice.http.HttpClient
+import org.lfdecentralizedtrust.splice.http.v0.definitions.GetBulkObjectChecksumsResponse
+import org.lfdecentralizedtrust.splice.scan.admin.api.client.{
+  BftScanConnection,
+  SingleScanConnection,
+}
 import org.lfdecentralizedtrust.splice.scan.config.BulkStorageConfig
+import org.lfdecentralizedtrust.splice.scan.store.{ScanInfo, ScanStore}
 import org.lfdecentralizedtrust.splice.store.S3BucketConnection.ObjectKeyAndChecksum
 import org.lfdecentralizedtrust.splice.store.{HasS3Mock, StoreTestBase}
 import org.lfdecentralizedtrust.splice.store.db.SplicePostgresTest
+import org.lfdecentralizedtrust.splice.util.TemplateJsonDecoder
 
 import java.security.MessageDigest
 import java.util.Base64
-import scala.concurrent.Future
+import scala.concurrent.{ExecutionContext, ExecutionContextExecutor, Future}
 import scala.jdk.CollectionConverters.*
+import scala.concurrent.duration.*
 
 class BulkStorageCommitFromStagingTest
     extends StoreTestBase
     with HasExecutionContext
     with HasActorSystem
     with HasS3Mock
-    with SplicePostgresTest {
+    with SplicePostgresTest
+    with HasRetryProvider {
 
   override val initialBuckets = Seq("staging", "committed")
-  val appConfig = BulkStorageConfig()
+  implicit val httpClient: HttpClient = null
+  implicit val templateJsonDecoder: TemplateJsonDecoder = null
 
   "BulkStorageCommitFromStaging" should {
+    val appConfig = BulkStorageConfig(
+      bftCheckEnabled = false
+    )
+
     "successfully move objects from staging to committed S3 bucket" in {
 
       val (stagingS3Connection, committedS3Connection, objsWithDigests) = setupTest
 
-      triggerCopyFlow(stagingS3Connection, committedS3Connection, objsWithDigests)
+      triggerCopyFlowAndAssertCompletion(
+        newCopyFlow(stagingS3Connection, committedS3Connection, objsWithDigests)
+      )
 
       assertObjectsMoved(stagingS3Connection, committedS3Connection, objsWithDigests)
     }
@@ -54,37 +82,303 @@ class BulkStorageCommitFromStagingTest
 
       loggerFactory.assertLogsSeq(SuppressionRule.LevelAndAbove(Level.DEBUG))(
         {
-          triggerCopyFlow(stagingS3Connection, committedS3Connection, objsWithDigests)
+          triggerCopyFlowAndAssertCompletion(
+            newCopyFlow(stagingS3Connection, committedS3Connection, objsWithDigests)
+          )
         },
         logEntries => forExactly(1, logEntries)(_.message should include("Skipping copy")),
       )
 
       assertObjectsMoved(stagingS3Connection, committedS3Connection, objsWithDigests)
     }
+
+    def newCopyFlow(
+        stagingS3Connection: S3BucketConnectionForUnitTests,
+        committedS3Connection: S3BucketConnectionForUnitTests,
+        objsWithDigests: Seq[ObjectKeyAndChecksum],
+    ) = {
+      BulkStorageCommitFromStaging[String](
+        stagingS3Connection,
+        committedS3Connection,
+        _ => Future.successful(objsWithDigests),
+        appConfig,
+        null, // not used when bft reads are disabled
+        null, // not used when bft reads are disabled
+        null, // not used when bft reads are disabled
+        null, // not used when bft reads are disabled
+        null, // not used when bft reads are disabled
+        null, // not used when bft reads are disabled
+        null, // not used when bft reads are disabled
+        loggerFactory,
+      )
+    }
+
   }
 
-  private def triggerCopyFlow(
-      stagingS3Connection: S3BucketConnectionForUnitTests,
-      committedS3Connection: S3BucketConnectionForUnitTests,
-      objsWithDigests: Seq[ObjectKeyAndChecksum],
-  ) = {
-    val flow = BulkStorageCommitFromStaging[String](
-      stagingS3Connection,
-      committedS3Connection,
-      _ => Future.successful(objsWithDigests),
-      appConfig,
-      loggerFactory,
+  "BulkStorageCommitFromStaging with BFT reads enabled" should {
+    val appConfig = BulkStorageConfig(
+      bftRetryInterval = NonNegativeFiniteDuration.ofSeconds(1)
     )
 
+    "successfully move objects from staging to committed S3 bucket when there's full consensus" in {
+      val (stagingS3Connection, committedS3Connection, objsWithDigests) = setupTest
+
+      val mockScanConnections = new MockScanConnections(objsWithDigests)
+      Seq.range(0, 7).foreach { i =>
+        mockScanConnections.scanAgrees(i)
+      }
+
+      val flow = newCopyFlow(
+        stagingS3Connection,
+        committedS3Connection,
+        objsWithDigests,
+        mockScanConnections,
+      )
+
+      triggerCopyFlowAndAssertCompletion(flow)
+
+      assertObjectsMoved(stagingS3Connection, committedS3Connection, objsWithDigests)
+    }
+
+    "wait until all objects are known to the peers, and report disagreement on consensus correctly" in {
+      val (stagingS3Connection, committedS3Connection, objsWithDigests) = setupTest
+
+      val mockScanConnections = new MockScanConnections(objsWithDigests)
+
+      val flow = newCopyFlow(
+        stagingS3Connection,
+        committedS3Connection,
+        objsWithDigests,
+        mockScanConnections,
+      )
+
+      val (pub, sub) = TestSource
+        .probe[String]
+        .via(flow)
+        .toMat(TestSink.probe[String])(Keep.both)
+        .run()
+
+      try {
+        clue("When one object is not known to the peers, the copy flow should not complete") {
+          Seq.range(0, 2).foreach(i => mockScanConnections.scanAgrees(i))
+          Seq.range(2, 7).foreach(i => mockScanConnections.scanMissingAnObject(i, 1))
+
+          sub.request(1)
+          pub.sendNext("go")
+          sub.expectNoMessage(20.seconds)
+
+          stagingS3Connection.listObjects.futureValue
+            .contents()
+            .asScala should have size objsWithDigests.size.toLong
+          committedS3Connection.listObjects.futureValue.contents().asScala shouldBe empty
+        }
+
+        // errors on mismatching digests continue past the first clue for some time until enough scans are updated to agree on the digests,
+        // so we make the assertion on the logs fairly wide here to avoid the late error logs failing the log checker
+        loggerFactory.assertLogsSeq(SuppressionRule.LevelAndAbove(Level.ERROR))(
+          {
+            clue(
+              "Simulate a majority disagreeing with our digests, the copy flow should not complete and an error should be emitted"
+            ) {
+              Seq.range(2, 7).foreach(i => mockScanConnections.scanDisagreesOnDigest(i, 1))
+              sub.expectNoMessage(20.seconds)
+              stagingS3Connection.listObjects.futureValue
+                .contents()
+                .asScala should have size objsWithDigests.size.toLong
+              committedS3Connection.listObjects.futureValue.contents().asScala shouldBe empty
+
+            }
+
+            clue("Enough scans do agree - the copy flow should complete successfully") {
+              Seq.range(2, 5).foreach(i => mockScanConnections.scanAgrees(i))
+              sub.expectNext(20.seconds, "go")
+              assertObjectsMoved(stagingS3Connection, committedS3Connection, objsWithDigests)
+            }
+
+          },
+          logEntries =>
+            forAtLeast(1, logEntries)(
+              _.message should include(
+                "All objects are known to the BFT peers, but the checksums do not match"
+              )
+            ),
+        )
+
+      } catch {
+        case ex: Throwable =>
+          pub.sendError(ex)
+          sub.cancel()
+          throw ex
+      }
+    }
+
+    class MockScanConnections(
+        objsWithDigests: Seq[ObjectKeyAndChecksum]
+    ) {
+
+      private val singleScanConnections: Seq[SingleScanConnection] = Seq.range(0, 7).map { i =>
+        val mockConn = mock[SingleScanConnection]
+        when(mockConn.config) thenReturn ScanAppClientConfig(
+          NetworkAppClientConfig(
+            Uri(s"http://dummy-admin-$i")
+          )
+        )
+        when(mockConn.url) thenReturn Uri(s"http://scan_$i")
+        mockConn
+      }
+
+      def scanAgrees(idx: Integer): Unit = {
+        when(
+          singleScanConnections(idx)
+            .getBulkObjectChecksums(any[Seq[String]])(any[ExecutionContext], any[TraceContext])
+        )
+          .thenReturn(
+            Future.successful(
+              new GetBulkObjectChecksumsResponse(
+                objsWithDigests
+                  .map(_.checksum)
+                  .map(digest => new GetBulkObjectChecksumsResponse.Checksums(Some(digest)))
+                  .toVector
+              )
+            )
+          )
+        ()
+      }
+
+      def scanDisagreesOnDigest(scanIdx: Integer, objIdx: Integer): Unit = {
+        when(
+          singleScanConnections(scanIdx)
+            .getBulkObjectChecksums(any[Seq[String]])(any[ExecutionContext], any[TraceContext])
+        )
+          .thenReturn(
+            Future.successful(
+              new GetBulkObjectChecksumsResponse(
+                objsWithDigests
+                  .map(_.checksum)
+                  .updated(objIdx, "wrong-digest")
+                  .map(digest => new GetBulkObjectChecksumsResponse.Checksums(Some(digest)))
+                  .toVector
+              )
+            )
+          )
+      }
+
+      def scanMissingAnObject(scanIdx: Integer, objIdx: Integer): Unit = {
+        when(
+          singleScanConnections(scanIdx)
+            .getBulkObjectChecksums(any[Seq[String]])(any[ExecutionContext], any[TraceContext])
+        )
+          .thenReturn(
+            Future.successful(
+              new GetBulkObjectChecksumsResponse(
+                objsWithDigests
+                  .map(_.checksum)
+                  .map(Some(_))
+                  .updated(objIdx, None)
+                  .map(oDigest => new GetBulkObjectChecksumsResponse.Checksums(oDigest))
+                  .toVector
+              )
+            )
+          )
+      }
+
+      private val scanList = new BftScanConnection.AllDsoScansBft(
+        initialScanConnections = singleScanConnections,
+        initialFailedConnections = Map.empty,
+        connectionBuilder = _ => Future.failed(new RuntimeException("Shouldn't be refreshing!")),
+        scanUrlsChangedCallback = _ => Future.unit,
+        getScans = BftScanConnection.Bft.getScansInDsoRules,
+        scansRefreshInterval = NonNegativeFiniteDuration.ofDays(10),
+        retryProvider = testRetryProvider,
+        loggerFactory = loggerFactory,
+      )
+      val bftConnection = new BftScanConnection(
+        amuletLedgerClient = mock[SpliceLedgerClient],
+        amuletRulesCacheTimeToLive = NonNegativeFiniteDuration.ofSeconds(1),
+        scanList = scanList,
+        clock = wallClock,
+        retryProvider = testRetryProvider,
+        loggerFactory = loggerFactory,
+      )
+      private val syncId = "dummy::dummy"
+      val scanStore: ScanStore = mock[ScanStore]
+      when(scanStore.getDecentralizedSynchronizerId()(any[TraceContext]))
+        .thenReturn(Future.successful(SynchronizerId.tryFromString(syncId)))
+      when(scanStore.listDsoScans()(any[TraceContext]))
+        .thenReturn(
+          Future.successful(
+            Seq(
+              syncId -> Seq
+                .range(0, 7)
+                .map(i => ScanInfo(s"https://dummy-admin-$i", s"sv-$i"))
+                .toVector
+            ).toVector
+          )
+        )
+
+    }
+
+    def newCopyFlow(
+        stagingS3Connection: S3BucketConnectionForUnitTests,
+        committedS3Connection: S3BucketConnectionForUnitTests,
+        objsWithDigests: Seq[ObjectKeyAndChecksum],
+        mockScanConnections: MockScanConnections,
+    ) = {
+      new BulkStorageCommitFromStaging[String](
+        stagingS3Connection,
+        committedS3Connection,
+        _ => Future.successful(objsWithDigests),
+        appConfig,
+        mockScanConnections.scanStore,
+        "sv-1",
+        null, // no ledger client, we're using real BFT with mock single connections
+        AutomationConfig(),
+        UpgradesConfig(),
+        wallClock,
+        testRetryProvider,
+        loggerFactory,
+      ) {
+        override protected def getOrCreateScanConnection(
+            store: ScanStore,
+            svName: String,
+            ledgerClient: SpliceLedgerClient,
+            automationConfig: AutomationConfig,
+            upgradesConfig: UpgradesConfig,
+            clock: Clock,
+            retryProvider: RetryProvider,
+            loggerFactory: NamedLoggerFactory,
+        )(implicit
+            tc: TraceContext,
+            ec: ExecutionContextExecutor,
+            mat: Materializer,
+            httpClient: HttpClient,
+            templateJsonDecoder: TemplateJsonDecoder,
+        ): Future[BftScanConnection] = Future.successful(mockScanConnections.bftConnection)
+      }.getFlow
+    }
+  }
+
+  private def triggerCopyFlowAndAssertCompletion(
+      flow: Flow[String, String, NotUsed]
+  ) = {
     val (pub, sub) = TestSource
       .probe[String]
       .via(flow)
       .toMat(TestSink.probe[String])(Keep.both)
       .run()
 
-    sub.request(1)
-    pub.sendNext("go")
-    sub.expectNext("go")
+    try {
+      sub.request(1)
+      pub.sendNext("go")
+      sub.expectNext("go")
+      pub.sendComplete()
+      sub.expectComplete()
+    } catch {
+      case ex: Throwable =>
+        pub.sendError(ex)
+        sub.cancel()
+        throw ex
+    }
   }
 
   private def assertObjectsMoved(
diff --git a/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/store/bulk/UpdateHistoryBulkStorageTest.scala b/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/store/bulk/UpdateHistoryBulkStorageTest.scala
index 62ddcd750e..b120843ace 100644
--- a/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/store/bulk/UpdateHistoryBulkStorageTest.scala
+++ b/apps/scan/src/test/scala/org/lfdecentralizedtrust/splice/scan/store/bulk/UpdateHistoryBulkStorageTest.scala
@@ -57,7 +57,8 @@ class UpdateHistoryBulkStorageTest
     zstdCompressionLevel = 3,
   )
   val appConfig = BulkStorageConfig(
-    updatesPollingInterval = NonNegativeFiniteDuration.ofSeconds(5)
+    updatesPollingInterval = NonNegativeFiniteDuration.ofSeconds(5),
+    bftCheckEnabled = false, // bft checks are tested elsewhere
   )
 
   "UpdateHistoryBulkStorage" should {
diff --git a/cluster/configs/shared/rate-limits/unlimited.yaml b/cluster/configs/shared/rate-limits/unlimited.yaml
index 65b069f8b8..fe11ffb440 100644
--- a/cluster/configs/shared/rate-limits/unlimited.yaml
+++ b/cluster/configs/shared/rate-limits/unlimited.yaml
@@ -185,6 +185,10 @@ rateLimits:
     name: listBulkUpdateHistoryObjects
     type: unlimited
 
+  /api/scan/v0/history/bulk/checksums:
+    name: getBulkHistoryChecksums
+    type: unlimited
+
   /api/scan/v0/active-synchronizer-serial:
     name: activeSynchronizerSerial
     type: unlimited
diff --git a/cluster/deployment/scratchneta/config.resolved.yaml b/cluster/deployment/scratchneta/config.resolved.yaml
index a01269d394..e7a1e2c8bb 100644
--- a/cluster/deployment/scratchneta/config.resolved.yaml
+++ b/cluster/deployment/scratchneta/config.resolved.yaml
@@ -287,6 +287,9 @@ sv:
         /api/scan/v0/history/bulk/acs:
           name: 'listBulkAcsSnapshotObjects'
           type: 'unlimited'
+        /api/scan/v0/history/bulk/checksums:
+          name: 'getBulkHistoryChecksums'
+          type: 'unlimited'
         /api/scan/v0/history/bulk/updates:
           name: 'listBulkUpdateHistoryObjects'
           type: 'unlimited'
diff --git a/cluster/deployment/scratchnetb/config.resolved.yaml b/cluster/deployment/scratchnetb/config.resolved.yaml
index a01269d394..e7a1e2c8bb 100644
--- a/cluster/deployment/scratchnetb/config.resolved.yaml
+++ b/cluster/deployment/scratchnetb/config.resolved.yaml
@@ -287,6 +287,9 @@ sv:
         /api/scan/v0/history/bulk/acs:
           name: 'listBulkAcsSnapshotObjects'
           type: 'unlimited'
+        /api/scan/v0/history/bulk/checksums:
+          name: 'getBulkHistoryChecksums'
+          type: 'unlimited'
         /api/scan/v0/history/bulk/updates:
           name: 'listBulkUpdateHistoryObjects'
           type: 'unlimited'
diff --git a/cluster/deployment/scratchnetc/config.resolved.yaml b/cluster/deployment/scratchnetc/config.resolved.yaml
index a01269d394..e7a1e2c8bb 100644
--- a/cluster/deployment/scratchnetc/config.resolved.yaml
+++ b/cluster/deployment/scratchnetc/config.resolved.yaml
@@ -287,6 +287,9 @@ sv:
         /api/scan/v0/history/bulk/acs:
           name: 'listBulkAcsSnapshotObjects'
           type: 'unlimited'
+        /api/scan/v0/history/bulk/checksums:
+          name: 'getBulkHistoryChecksums'
+          type: 'unlimited'
         /api/scan/v0/history/bulk/updates:
           name: 'listBulkUpdateHistoryObjects'
           type: 'unlimited'
diff --git a/cluster/deployment/scratchnetd/config.resolved.yaml b/cluster/deployment/scratchnetd/config.resolved.yaml
index a01269d394..e7a1e2c8bb 100644
--- a/cluster/deployment/scratchnetd/config.resolved.yaml
+++ b/cluster/deployment/scratchnetd/config.resolved.yaml
@@ -287,6 +287,9 @@ sv:
         /api/scan/v0/history/bulk/acs:
           name: 'listBulkAcsSnapshotObjects'
           type: 'unlimited'
+        /api/scan/v0/history/bulk/checksums:
+          name: 'getBulkHistoryChecksums'
+          type: 'unlimited'
         /api/scan/v0/history/bulk/updates:
           name: 'listBulkUpdateHistoryObjects'
           type: 'unlimited'
diff --git a/cluster/deployment/scratchnete/config.resolved.yaml b/cluster/deployment/scratchnete/config.resolved.yaml
index a01269d394..e7a1e2c8bb 100644
--- a/cluster/deployment/scratchnete/config.resolved.yaml
+++ b/cluster/deployment/scratchnete/config.resolved.yaml
@@ -287,6 +287,9 @@ sv:
         /api/scan/v0/history/bulk/acs:
           name: 'listBulkAcsSnapshotObjects'
           type: 'unlimited'
+        /api/scan/v0/history/bulk/checksums:
+          name: 'getBulkHistoryChecksums'
+          type: 'unlimited'
         /api/scan/v0/history/bulk/updates:
           name: 'listBulkUpdateHistoryObjects'
           type: 'unlimited'
diff --git a/cluster/expected/canton-network/expected.json b/cluster/expected/canton-network/expected.json
index ce454feac0..c7b841e7eb 100644
--- a/cluster/expected/canton-network/expected.json
+++ b/cluster/expected/canton-network/expected.json
@@ -1720,6 +1720,10 @@
           "name": "listBulkAcsSnapshotObjects",
           "type": "unlimited"
         },
+        "/api/scan/v0/history/bulk/checksums": {
+          "name": "getBulkHistoryChecksums",
+          "type": "unlimited"
+        },
         "/api/scan/v0/history/bulk/updates": {
           "name": "listBulkUpdateHistoryObjects",
           "type": "unlimited"
@@ -2103,6 +2107,10 @@
           "name": "listBulkAcsSnapshotObjects",
           "type": "unlimited"
         },
+        "/api/scan/v0/history/bulk/checksums": {
+          "name": "getBulkHistoryChecksums",
+          "type": "unlimited"
+        },
         "/api/scan/v0/history/bulk/updates": {
           "name": "listBulkUpdateHistoryObjects",
           "type": "unlimited"
diff --git a/cluster/expected/sv-runbook/expected.json b/cluster/expected/sv-runbook/expected.json
index a149bccb94..a833c57f20 100644
--- a/cluster/expected/sv-runbook/expected.json
+++ b/cluster/expected/sv-runbook/expected.json
@@ -945,6 +945,10 @@
           "name": "listBulkAcsSnapshotObjects",
           "type": "unlimited"
         },
+        "/api/scan/v0/history/bulk/checksums": {
+          "name": "getBulkHistoryChecksums",
+          "type": "unlimited"
+        },
         "/api/scan/v0/history/bulk/updates": {
           "name": "listBulkUpdateHistoryObjects",
           "type": "unlimited"