From 439458b0bc509c8bd4c62bf62db4f09c031e66a5 Mon Sep 17 00:00:00 2001 From: Itai Segall Date: Wed, 15 Jul 2026 22:05:55 +0000 Subject: [PATCH 01/17] [ci] add API to get object checksums Signed-off-by: Itai Segall --- .../splice/console/ScanAppReference.scala | 10 ++ apps/scan/src/main/openapi/scan.yaml | 48 ++++++++ .../client/commands/HttpScanAppClient.scala | 109 ++++++------------ .../scan/admin/http/HttpScanHandler.scala | 107 +++++------------ .../scan/store/bulk/BulkStorageReader.scala | 16 +++ 5 files changed, 143 insertions(+), 147 deletions(-) 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..e8dff15cf2 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,16 @@ 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/scan/src/main/openapi/scan.yaml b/apps/scan/src/main/openapi/scan.yaml index 527fe5fceb..7f34ce4885 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,30 @@ 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: string + BulkStorageObjectRef: type: object required: 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..1b4d2e1231 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 @@ -12,91 +12,32 @@ import com.daml.tls.TlsClientConfig import com.digitalasset.canton.config.RequireTypes import com.digitalasset.canton.config.RequireTypes.NonNegativeInt import org.lfdecentralizedtrust.splice.admin.api.client.commands.HttpCommand -import org.lfdecentralizedtrust.splice.codegen.java.splice.amulet.{ - FeaturedAppRight, - UnclaimedDevelopmentFundCoupon, -} -import org.lfdecentralizedtrust.splice.codegen.java.splice.amuletrules.{ - AmuletRules, - AppTransferContext, - TransferPreapproval, -} -import org.lfdecentralizedtrust.splice.codegen.java.splice.externalpartyamuletrules.{ - ExternalPartyAmuletRules, - TransferCommandCounter, -} -import org.lfdecentralizedtrust.splice.codegen.java.splice.round.{ - ClosedMiningRound, - IssuingMiningRound, - OpenMiningRound, -} +import org.lfdecentralizedtrust.splice.codegen.java.splice.amulet.{FeaturedAppRight, UnclaimedDevelopmentFundCoupon} +import org.lfdecentralizedtrust.splice.codegen.java.splice.amuletrules.{AmuletRules, AppTransferContext, TransferPreapproval} +import org.lfdecentralizedtrust.splice.codegen.java.splice.externalpartyamuletrules.{ExternalPartyAmuletRules, TransferCommandCounter} +import org.lfdecentralizedtrust.splice.codegen.java.splice.round.{ClosedMiningRound, IssuingMiningRound, OpenMiningRound} import org.lfdecentralizedtrust.splice.codegen.java.splice.ans as ansCodegen import org.lfdecentralizedtrust.splice.codegen.java.splice.ans.AnsRules import org.lfdecentralizedtrust.splice.config.SpliceInstanceNamesConfig import org.lfdecentralizedtrust.splice.http.v0.{definitions, scan as http} -import org.lfdecentralizedtrust.tokenstandard.{ - allocation, - allocationinstruction, - metadata, - transferinstruction, -} -import org.lfdecentralizedtrust.splice.http.v0.scan.{ - ForceAcsSnapshotNowResponse, - GetDateOfFirstSnapshotAfterResponse, - GetDateOfMostRecentSnapshotBeforeResponse, - GetLsuResponse, - ListBulkAcsSnapshotObjectsResponse, - ListBulkUpdateHistoryObjectsResponse, -} -import org.lfdecentralizedtrust.splice.scan.admin.http.{ - CompactJsonScanHttpEncodings, - ProtobufJsonScanHttpEncodings, -} +import org.lfdecentralizedtrust.tokenstandard.{allocation, allocationinstruction, metadata, transferinstruction} +import org.lfdecentralizedtrust.splice.http.v0.scan.{ForceAcsSnapshotNowResponse, GetBulkObjectChecksumsResponse, GetDateOfFirstSnapshotAfterResponse, GetDateOfMostRecentSnapshotBeforeResponse, GetLsuResponse, ListBulkAcsSnapshotObjectsResponse, ListBulkUpdateHistoryObjectsResponse} +import org.lfdecentralizedtrust.splice.scan.admin.http.{CompactJsonScanHttpEncodings, ProtobufJsonScanHttpEncodings} import org.lfdecentralizedtrust.splice.store.HistoryBackfilling.SourceMigrationInfo import org.lfdecentralizedtrust.splice.store.{MultiDomainAcsStore, VoteResultsFilters} import org.lfdecentralizedtrust.splice.store.UpdateHistory.UpdateHistoryResponse -import org.lfdecentralizedtrust.splice.util.{ - ChoiceContextWithDisclosures, - Codec, - Contract, - ContractWithState, - DomainRecordTimeRange, - FactoryChoiceWithDisclosures, - PackageQualifiedName, - TemplateJsonDecoder, -} +import org.lfdecentralizedtrust.splice.util.{ChoiceContextWithDisclosures, Codec, Contract, ContractWithState, DomainRecordTimeRange, FactoryChoiceWithDisclosures, PackageQualifiedName, TemplateJsonDecoder} import com.digitalasset.canton.data.CantonTimestamp import com.digitalasset.canton.logging.pretty.{Pretty, PrettyPrinting} import com.digitalasset.canton.synchronizer.sequencer.block.bftordering.bindings.p2p.grpc.P2PGrpcNetworking.P2PEndpoint import com.digitalasset.canton.synchronizer.sequencer.block.bftordering.core.BftBlockOrdererConfig.P2PEndpointConfig -import com.digitalasset.canton.topology.{ - Member, - ParticipantId, - PartyId, - PhysicalSynchronizerId, - SequencerId, - SynchronizerId, -} +import com.digitalasset.canton.topology.{Member, ParticipantId, PartyId, PhysicalSynchronizerId, SequencerId, SynchronizerId} import com.digitalasset.daml.lf.data.Time.Timestamp import com.google.protobuf.ByteString import org.apache.pekko.stream.scaladsl.Source -import org.lfdecentralizedtrust.splice.codegen.java.splice.api.token.{ - allocationinstructionv1, - allocationinstructionv2, - allocationv1, - allocationv2, - metadatav1, - transferinstructionv1, - transferinstructionv2, -} -import org.lfdecentralizedtrust.splice.codegen.java.splice.dsorules.{ - DsoRules_CloseVoteRequestResult, - VoteRequest, -} -import org.lfdecentralizedtrust.splice.scan.admin.api.client.{ - BulkStorageDownloadResponse, - ScanStreamClient, -} +import org.lfdecentralizedtrust.splice.codegen.java.splice.api.token.{allocationinstructionv1, allocationinstructionv2, allocationv1, allocationv2, metadatav1, transferinstructionv1, transferinstructionv2} +import org.lfdecentralizedtrust.splice.codegen.java.splice.dsorules.{DsoRules_CloseVoteRequestResult, VoteRequest} +import org.lfdecentralizedtrust.splice.scan.admin.api.client.{BulkStorageDownloadResponse, ScanStreamClient} import java.util.Base64 import java.time.Instant @@ -3371,6 +3312,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..9747ae98fb 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 @@ -16,14 +16,7 @@ import com.digitalasset.canton.time.Clock import com.digitalasset.canton.topology.store.TimeQuery import com.digitalasset.canton.topology.{Member, PartyId, SynchronizerId} import com.digitalasset.canton.tracing.TraceContext -import com.digitalasset.canton.util.{ - ByteStringUtil, - ErrorUtil, - GrpcStreamingUtils, - MaxBytesToDecompress, - MonadUtil, - ResourceUtil, -} +import com.digitalasset.canton.util.{ByteStringUtil, ErrorUtil, GrpcStreamingUtils, MaxBytesToDecompress, MonadUtil, ResourceUtil} import com.digitalasset.canton.util.ShowUtil.* import com.digitalasset.daml.lf.data.Time.Timestamp import com.github.blemale.scaffeine.{Cache, Scaffeine} @@ -36,81 +29,28 @@ import org.lfdecentralizedtrust.splice.codegen.java.splice.{amulet, ans as ansCo import org.lfdecentralizedtrust.splice.codegen.java.splice.amuletrules.AmuletRules import org.lfdecentralizedtrust.splice.codegen.java.splice.dso.decentralizedsynchronizer.SynchronizerNodeConfig import org.lfdecentralizedtrust.splice.codegen.java.splice.dso.svstate.SvNodeState -import org.lfdecentralizedtrust.splice.codegen.java.splice.externalpartyamuletrules.{ - ExternalPartyAmuletRules, - TransferCommand, -} -import org.lfdecentralizedtrust.splice.codegen.java.splice.round.{ - ClosedMiningRound, - IssuingMiningRound, - OpenMiningRound, - SummarizingMiningRound, -} +import org.lfdecentralizedtrust.splice.codegen.java.splice.externalpartyamuletrules.{ExternalPartyAmuletRules, TransferCommand} +import org.lfdecentralizedtrust.splice.codegen.java.splice.round.{ClosedMiningRound, IssuingMiningRound, OpenMiningRound, SummarizingMiningRound} import org.lfdecentralizedtrust.splice.config.{SpliceInstanceNamesConfig, Thresholds} -import org.lfdecentralizedtrust.splice.environment.{ - PackageVersionSupport, - ParticipantAdminConnection, - SequencerAdminConnection, - SynchronizerNodeService, -} -import org.lfdecentralizedtrust.splice.environment.TopologyAdminConnection.{ - TopologySnapshot, - TopologyTransactionType, -} +import org.lfdecentralizedtrust.splice.environment.{PackageVersionSupport, ParticipantAdminConnection, SequencerAdminConnection, SynchronizerNodeService} +import org.lfdecentralizedtrust.splice.environment.TopologyAdminConnection.{TopologySnapshot, TopologyTransactionType} import org.lfdecentralizedtrust.splice.environment.TopologyAdminConnection.TopologyTransactionType.AuthorizedState -import org.lfdecentralizedtrust.splice.http.{ - HttpFeatureSupportHandler, - HttpValidatorLicensesHandler, - HttpVotesHandler, - UrlValidator, -} +import org.lfdecentralizedtrust.splice.http.{HttpFeatureSupportHandler, HttpValidatorLicensesHandler, HttpVotesHandler, UrlValidator} import org.lfdecentralizedtrust.splice.http.v0.{definitions, scan as v0} -import org.lfdecentralizedtrust.splice.http.v0.definitions.{ - AcsRequest, - BatchListVotesByVoteRequestsRequest, - DamlValueEncoding, - CountVoteResultsRequest, - ErrorResponse, - EventHistoryRequest, - HoldingsStateRequest, - HoldingsSummaryRequest, - HoldingsSummaryRequestV1, - ListBulkUpdateHistoryObjectsRequest, - ListVoteResultsRequest, - MaybeCachedContractWithState, - PreviousSvRewardWeightRequest, - PreviousSvRewardWeightResponse, - UpdateHistoryItem, - UpdateHistoryItemV2WithHash, - UpdateHistoryRequestV2, - UpdateHistoryTransactionV2WithHash, -} +import org.lfdecentralizedtrust.splice.http.v0.definitions.{AcsRequest, BatchListVotesByVoteRequestsRequest, CountVoteResultsRequest, DamlValueEncoding, ErrorResponse, EventHistoryRequest, GetBulkObjectChecksumsRequest, HoldingsStateRequest, HoldingsSummaryRequest, HoldingsSummaryRequestV1, ListBulkUpdateHistoryObjectsRequest, ListVoteResultsRequest, MaybeCachedContractWithState, PreviousSvRewardWeightRequest, PreviousSvRewardWeightResponse, UpdateHistoryItem, UpdateHistoryItemV2WithHash, UpdateHistoryRequestV2, UpdateHistoryTransactionV2WithHash} import org.lfdecentralizedtrust.splice.http.v0.scan.ScanResource import org.lfdecentralizedtrust.splice.scan.ScanSynchronizerNode import org.lfdecentralizedtrust.splice.scan.admin.http.ScanHttpEncodings.updateV1ToUpdateV2 import org.lfdecentralizedtrust.splice.scan.config.{CantonBftPeerConfig, ScanRollForwardLsuConfig} import org.lfdecentralizedtrust.splice.scan.dso.DsoAnsResolver -import org.lfdecentralizedtrust.splice.scan.store.{ - AcsSnapshotStore, - AppActivityStore, - ScanEventStore, - ScanStore, - TxLogEntry, -} +import org.lfdecentralizedtrust.splice.scan.store.{AcsSnapshotStore, AppActivityStore, ScanEventStore, ScanStore, TxLogEntry} import org.lfdecentralizedtrust.splice.scan.store.bulk.BulkStorageReader import org.lfdecentralizedtrust.splice.scan.store.AcsSnapshotStore.QueryAcsSnapshotResult import org.lfdecentralizedtrust.splice.scan.store.bulk.AcsSnapshotBulkStorage.AcsSnapshotObjects import org.lfdecentralizedtrust.splice.scan.store.bulk.UpdateHistoryBulkStorage.UpdateHistoryObjectsResponse import org.lfdecentralizedtrust.splice.store.AppStoreWithIngestion.SpliceLedgerConnectionPriority import org.lfdecentralizedtrust.splice.store.MultiDomainAcsStore.TxLogBackfillingState -import org.lfdecentralizedtrust.splice.store.{ - AppStore, - AppStoreWithIngestion, - PageLimit, - SortOrder, - VoteResultsFilters, - VotesStore, -} +import org.lfdecentralizedtrust.splice.store.{AppStore, AppStoreWithIngestion, PageLimit, SortOrder, VoteResultsFilters, VotesStore} import org.lfdecentralizedtrust.splice.store.S3BucketConnection.ObjectKeyAndChecksum import org.lfdecentralizedtrust.splice.store.UpdateHistory.BackfillingState import org.lfdecentralizedtrust.splice.store.UpdateHistory @@ -118,13 +58,7 @@ import org.lfdecentralizedtrust.splice.store.UpdateHistory import java.lang.IllegalStateException import scala.collection.immutable.SortedMap import org.lfdecentralizedtrust.splice.scan.store.db.DbScanAppRewardsStore -import org.lfdecentralizedtrust.splice.util.{ - Codec, - Contract, - ContractWithState, - PackageQualifiedName, - QualifiedName, -} +import org.lfdecentralizedtrust.splice.util.{Codec, Contract, ContractWithState, PackageQualifiedName, QualifiedName} import org.lfdecentralizedtrust.splice.util.PrettyInstances.* import java.io.ByteArrayInputStream @@ -2649,6 +2583,27 @@ 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.toVector + ) + ) + } + } + } + } + def getRollForwardLsu(respond: ScanResource.GetRollForwardLsuResponse.type)()( extracted: TraceContext ): Future[ScanResource.GetRollForwardLsuResponse] = { 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..9ca07b5015 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,22 @@ class BulkStorageReader( } } + def getObjectChecksums( + objectKeys: Seq[String] + ): Future[Seq[String]] = { + for { + committed <- committedS3Connection.getChecksums(objectKeys) + staging <- stagingS3Connection.getChecksums(objectKeys) + } yield { + objectKeys.map { key => + committed.find(_.key == key).orElse(staging.find(_.key == key)) match { + case Some(obj) => obj.checksum + case None => "" + } + } + } + } + private def getSegmentStartingAt( startTimestamp: Option[CantonTimestamp] ): Future[Option[(CantonTimestamp, CantonTimestamp)]] = From 4fcf4b60ff7b3d7a1506724fa040269e7b47efe3 Mon Sep 17 00:00:00 2001 From: Itai Segall Date: Thu, 16 Jul 2026 20:07:37 +0000 Subject: [PATCH 02/17] [ci] scanConnection Signed-off-by: Itai Segall --- .../admin/api/client/BftScanConnection.scala | 6 ++++ .../admin/api/client/ScanConnection.scala | 36 +++++-------------- .../api/client/SingleScanConnection.scala | 9 +++++ 3 files changed, 24 insertions(+), 27 deletions(-) diff --git a/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/admin/api/client/BftScanConnection.scala b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/admin/api/client/BftScanConnection.scala index 3c9c689cda..322e9036b5 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,11 @@ class BftScanConnection( ) .transform(tryBatch => Success(tryBatch.toOption)) } + + override def getBulkObjectChecksums( + objectKeys: Seq[String] + )(implicit ec: ExecutionContext, tc: TraceContext): Future[GetBulkObjectChecksumsResponse] = + bftCall(_.getBulkObjectChecksums(objectKeys), "getBulkObjectChecksums") } 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..35659aaf2e 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 @@ -4,35 +4,17 @@ package org.lfdecentralizedtrust.splice.scan.admin.api.client import cats.data.OptionT -import org.lfdecentralizedtrust.splice.codegen.java.splice.amulet.{ - FeaturedAppRight, - UnclaimedDevelopmentFundCoupon, -} +import org.lfdecentralizedtrust.splice.codegen.java.splice.amulet.{FeaturedAppRight, UnclaimedDevelopmentFundCoupon} import org.lfdecentralizedtrust.splice.codegen.java.splice.amuletrules.* import org.lfdecentralizedtrust.splice.codegen.java.splice.ans.AnsRules import org.lfdecentralizedtrust.splice.codegen.java.splice.dsorules.DsoRules -import org.lfdecentralizedtrust.splice.codegen.java.splice.externalpartyamuletrules.{ - ExternalPartyAmuletRules, - TransferCommandCounter, -} -import org.lfdecentralizedtrust.splice.codegen.java.splice.round.{ - IssuingMiningRound, - OpenMiningRound, -} +import org.lfdecentralizedtrust.splice.codegen.java.splice.externalpartyamuletrules.{ExternalPartyAmuletRules, TransferCommandCounter} +import org.lfdecentralizedtrust.splice.codegen.java.splice.round.{IssuingMiningRound, OpenMiningRound} import org.lfdecentralizedtrust.splice.codegen.java.splice.types.Round import org.lfdecentralizedtrust.splice.config.UpgradesConfig import org.lfdecentralizedtrust.splice.environment.* import org.lfdecentralizedtrust.splice.http.HttpClient -import org.lfdecentralizedtrust.splice.http.v0.definitions.{ - GetDsoInfoResponse, - GetRewardAccountingActivityTotalsResponse, - GetRewardAccountingBatchResponse, - GetRewardAccountingRootHashResponse, - HoldingsSummaryResponse, - HoldingsSummaryResponseV1, - LookupTransferCommandStatusResponse, - MigrationSchedule, -} +import org.lfdecentralizedtrust.splice.http.v0.definitions.{GetBulkObjectChecksumsResponse, GetDsoInfoResponse, GetRewardAccountingActivityTotalsResponse, GetRewardAccountingBatchResponse, GetRewardAccountingRootHashResponse, HoldingsSummaryRequestV1, HoldingsSummaryResponse, HoldingsSummaryResponseV1, LookupTransferCommandStatusResponse, MigrationSchedule} import org.lfdecentralizedtrust.splice.scan.admin.api.client.ScanConnection.* import org.lfdecentralizedtrust.splice.scan.admin.api.client.commands.HttpScanAppClient import org.lfdecentralizedtrust.splice.scan.admin.api.client.commands.HttpScanAppClient.TransferContextWithInstances @@ -49,13 +31,9 @@ import com.digitalasset.canton.topology.{ParticipantId, PartyId, SynchronizerId} import com.digitalasset.canton.tracing.TraceContext import io.grpc.Status import org.apache.pekko.stream.Materializer -import org.lfdecentralizedtrust.splice.codegen.java.splice.dsorules.{ - DsoRules_CloseVoteRequestResult, - VoteRequest, -} +import org.lfdecentralizedtrust.splice.codegen.java.splice.dsorules.{DsoRules_CloseVoteRequestResult, VoteRequest} 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 +338,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..86bdb4de4c 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, @@ -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 { From 33c1927acb3a2fe3c2e908ece65b39a8b651d549 Mon Sep 17 00:00:00 2001 From: Itai Segall Date: Thu, 16 Jul 2026 20:31:31 +0000 Subject: [PATCH 03/17] [ci] refactor Signed-off-by: Itai Segall --- .../ScanHistoryBackfillingTrigger.scala | 45 ++++---------- .../scan/util/HasPeerBftScanConnection.scala | 58 +++++++++++++++++++ 2 files changed, 69 insertions(+), 34 deletions(-) create mode 100644 apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/util/HasPeerBftScanConnection.scala 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..86b5f50e03 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 @@ -41,6 +41,7 @@ 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 +62,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 +81,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 +220,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 { @@ -273,7 +243,14 @@ class ScanHistoryBackfillingTrigger( } private def performBackfilling()(implicit traceContext: TraceContext): Future[TaskOutcome] = for { - connection <- getOrCreateScanConnection() + connection <- getOrCreateScanConnection( + store, + svName, + ledgerClient, + context, + upgradesConfig, + loggerFactory, + ) backfilling = getOrCreateBackfilling(connection) outcome <- backfilling.backfill().map { case HistoryBackfilling.Outcome.MoreWorkAvailableNow(workDone) => 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..d5fb5eecf1 --- /dev/null +++ b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/util/HasPeerBftScanConnection.scala @@ -0,0 +1,58 @@ +package org.lfdecentralizedtrust.splice.scan.util + +import com.digitalasset.canton.logging.NamedLoggerFactory +import com.digitalasset.canton.tracing.TraceContext +import com.digitalasset.canton.util.Mutex +import org.lfdecentralizedtrust.splice.automation.TriggerContext +import org.lfdecentralizedtrust.splice.config.UpgradesConfig +import org.lfdecentralizedtrust.splice.environment.SpliceLedgerClient +import org.lfdecentralizedtrust.splice.scan.admin.api.client.BftScanConnection +import org.lfdecentralizedtrust.splice.scan.config.ScanAppClientConfig +import org.lfdecentralizedtrust.splice.scan.store.ScanStore + +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 + + def getOrCreateScanConnection( + store: ScanStore, + svName: String, + ledgerClient: SpliceLedgerClient, + context: TriggerContext, + upgradesConfig: UpgradesConfig, + loggerFactory: NamedLoggerFactory, + )(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 + } + } + } + } + +} From 9dc1a418b2386265fc9618a89378ff9d35615ea9 Mon Sep 17 00:00:00 2001 From: Itai Segall Date: Thu, 16 Jul 2026 20:56:19 +0000 Subject: [PATCH 04/17] [ci] refactor Signed-off-by: Itai Segall --- .../ScanHistoryBackfillingTrigger.scala | 26 ++++++++----------- .../scan/util/HasPeerBftScanConnection.scala | 23 ++++++++++++++-- 2 files changed, 32 insertions(+), 17 deletions(-) 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 86b5f50e03..7c304ab818 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 @@ -14,11 +14,7 @@ import org.lfdecentralizedtrust.splice.automation.{ import org.lfdecentralizedtrust.splice.config.UpgradesConfig import org.lfdecentralizedtrust.splice.environment.SpliceLedgerClient import org.lfdecentralizedtrust.splice.http.HttpClient -import org.lfdecentralizedtrust.splice.scan.admin.api.client.{ - BackfillingScanConnection, - BftScanConnection, -} -import org.lfdecentralizedtrust.splice.scan.config.ScanAppClientConfig +import org.lfdecentralizedtrust.splice.scan.admin.api.client.BackfillingScanConnection import org.lfdecentralizedtrust.splice.scan.store.ScanHistoryBackfilling.{ FoundingTransactionTreeUpdate, InitialTransactionTreeUpdate, @@ -35,7 +31,7 @@ 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 @@ -282,7 +278,14 @@ class ScanHistoryBackfillingTrigger( private def performImportUpdatesBackfilling()(implicit traceContext: TraceContext ): Future[TaskOutcome] = for { - connection <- getOrCreateScanConnection() + connection <- getOrCreateScanConnection( + store, + svName, + ledgerClient, + context, + upgradesConfig, + loggerFactory, + ) backfilling = getOrCreateBackfilling(connection) outcome <- backfilling.backfillImportUpdates().map { case ImportUpdatesBackfilling.Outcome.MoreWorkAvailableNow(workDone) => @@ -304,14 +307,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/util/HasPeerBftScanConnection.scala b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/util/HasPeerBftScanConnection.scala index d5fb5eecf1..62d1fcc10f 100644 --- 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 @@ -1,14 +1,18 @@ package org.lfdecentralizedtrust.splice.scan.util +import com.digitalasset.canton.lifecycle.SyncCloseable import com.digitalasset.canton.logging.NamedLoggerFactory import com.digitalasset.canton.tracing.TraceContext import com.digitalasset.canton.util.Mutex +import org.apache.pekko.stream.Materializer import org.lfdecentralizedtrust.splice.automation.TriggerContext import org.lfdecentralizedtrust.splice.config.UpgradesConfig import org.lfdecentralizedtrust.splice.environment.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} @@ -20,14 +24,20 @@ trait HasPeerBftScanConnection { @volatile private var connectionVar: Option[BftScanConnection] = None - def getOrCreateScanConnection( + protected def getOrCreateScanConnection( store: ScanStore, svName: String, ledgerClient: SpliceLedgerClient, context: TriggerContext, upgradesConfig: UpgradesConfig, loggerFactory: NamedLoggerFactory, - )(implicit tc: TraceContext): Future[BftScanConnection] = + )(implicit + tc: TraceContext, + ec: ExecutionContextExecutor, + mat: Materializer, + httpClient: HttpClient, + templateJsonDecoder: TemplateJsonDecoder, + ): Future[BftScanConnection] = blocking { mutex.exclusive { connectionVar match { @@ -55,4 +65,13 @@ trait HasPeerBftScanConnection { } } + protected def closeScanConnection(): Option[SyncCloseable] = + connectionVar + .map(connection => + SyncCloseable( + "closing scan connection", + connection.close(), + ) + ) + } From 00b8628cdd9f25bbf1e85f96c6238895143515ce Mon Sep 17 00:00:00 2001 From: Itai Segall Date: Thu, 16 Jul 2026 21:03:43 +0000 Subject: [PATCH 05/17] [ci] refactor Signed-off-by: Itai Segall --- .../ScanHistoryBackfillingTrigger.scala | 40 +++++++++++-------- .../scan/util/HasPeerBftScanConnection.scala | 16 ++++---- 2 files changed, 32 insertions(+), 24 deletions(-) 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 7c304ab818..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 @@ -14,7 +14,10 @@ import org.lfdecentralizedtrust.splice.automation.{ import org.lfdecentralizedtrust.splice.config.UpgradesConfig import org.lfdecentralizedtrust.splice.environment.SpliceLedgerClient import org.lfdecentralizedtrust.splice.http.HttpClient -import org.lfdecentralizedtrust.splice.scan.admin.api.client.BackfillingScanConnection +import org.lfdecentralizedtrust.splice.scan.admin.api.client.{ + BackfillingScanConnection, + BftScanConnection, +} import org.lfdecentralizedtrust.splice.scan.store.ScanHistoryBackfilling.{ FoundingTransactionTreeUpdate, InitialTransactionTreeUpdate, @@ -238,15 +241,25 @@ 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( - store, - svName, - ledgerClient, - context, - upgradesConfig, - loggerFactory, - ) + connection <- getOrCreateScanConnection() backfilling = getOrCreateBackfilling(connection) outcome <- backfilling.backfill().map { case HistoryBackfilling.Outcome.MoreWorkAvailableNow(workDone) => @@ -278,14 +291,7 @@ class ScanHistoryBackfillingTrigger( private def performImportUpdatesBackfilling()(implicit traceContext: TraceContext ): Future[TaskOutcome] = for { - connection <- getOrCreateScanConnection( - store, - svName, - ledgerClient, - context, - upgradesConfig, - loggerFactory, - ) + connection <- getOrCreateScanConnection() backfilling = getOrCreateBackfilling(connection) outcome <- backfilling.backfillImportUpdates().map { case ImportUpdatesBackfilling.Outcome.MoreWorkAvailableNow(workDone) => 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 index 62d1fcc10f..ad6f5550c6 100644 --- 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 @@ -2,12 +2,12 @@ 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.automation.TriggerContext -import org.lfdecentralizedtrust.splice.config.UpgradesConfig -import org.lfdecentralizedtrust.splice.environment.SpliceLedgerClient +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 @@ -28,8 +28,10 @@ trait HasPeerBftScanConnection { store: ScanStore, svName: String, ledgerClient: SpliceLedgerClient, - context: TriggerContext, + automationConfig: AutomationConfig, upgradesConfig: UpgradesConfig, + clock: Clock, + retryProvider: RetryProvider, loggerFactory: NamedLoggerFactory, )(implicit tc: TraceContext, @@ -50,11 +52,11 @@ trait HasPeerBftScanConnection { 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, + scansRefreshInterval = automationConfig.pollingInterval, amuletRulesCacheTimeToLive = ScanAppClientConfig.DefaultAmuletRulesCacheTimeToLive, upgradesConfig, - context.clock, - context.retryProvider, + clock, + retryProvider, loggerFactory, ) } yield { From eaa87d6b9be99d6a1a2e71178d728374b1cf8e96 Mon Sep 17 00:00:00 2001 From: Itai Segall Date: Fri, 17 Jul 2026 14:55:04 +0000 Subject: [PATCH 06/17] [ci] compiles Signed-off-by: Itai Segall --- .../splice/console/ScanAppReference.scala | 4 +- .../splice/scan/ScanApp.scala | 4 + .../admin/api/client/BftScanConnection.scala | 6 +- .../admin/api/client/ScanConnection.scala | 33 ++++++- .../client/commands/HttpScanAppClient.scala | 84 ++++++++++++++--- .../scan/admin/http/HttpScanHandler.scala | 91 ++++++++++++++++--- .../splice/scan/config/ScanAppConfig.scala | 1 + ...SnapshotBulkStorageCommitFromStaging.scala | 29 +++++- .../splice/scan/store/bulk/BulkStorage.scala | 48 ++++++++-- .../bulk/BulkStorageCommitFromStaging.scala | 81 ++++++++++++++++- ...eHistoryBulkStorageCommitFromStaging.scala | 30 +++++- .../scan/util/HasPeerBftScanConnection.scala | 3 + ...shotBulkStorageCommitFromStagingTest.scala | 15 ++- .../BulkStorageCommitFromStagingTest.scala | 16 +++- .../bulk/UpdateHistoryBulkStorageTest.scala | 3 +- 15 files changed, 398 insertions(+), 50 deletions(-) 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 e8dff15cf2..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,7 +908,9 @@ abstract class ScanAppReference( ) } - @Help.Summary("Get checksums for a list of bulk storage objects (using both staging and committed objects)") + @Help.Summary( + "Get checksums for a list of bulk storage objects (using both staging and committed objects)" + ) def getBulkObjectChecksums( objectKeys: Seq[String] ): definitions.GetBulkObjectChecksumsResponse = 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 322e9036b5..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 @@ -1125,7 +1125,11 @@ class BftScanConnection( override def getBulkObjectChecksums( objectKeys: Seq[String] )(implicit ec: ExecutionContext, tc: TraceContext): Future[GetBulkObjectChecksumsResponse] = - bftCall(_.getBulkObjectChecksums(objectKeys), "getBulkObjectChecksums") + 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 35659aaf2e..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 @@ -4,17 +4,37 @@ package org.lfdecentralizedtrust.splice.scan.admin.api.client import cats.data.OptionT -import org.lfdecentralizedtrust.splice.codegen.java.splice.amulet.{FeaturedAppRight, UnclaimedDevelopmentFundCoupon} +import org.lfdecentralizedtrust.splice.codegen.java.splice.amulet.{ + FeaturedAppRight, + UnclaimedDevelopmentFundCoupon, +} import org.lfdecentralizedtrust.splice.codegen.java.splice.amuletrules.* import org.lfdecentralizedtrust.splice.codegen.java.splice.ans.AnsRules import org.lfdecentralizedtrust.splice.codegen.java.splice.dsorules.DsoRules -import org.lfdecentralizedtrust.splice.codegen.java.splice.externalpartyamuletrules.{ExternalPartyAmuletRules, TransferCommandCounter} -import org.lfdecentralizedtrust.splice.codegen.java.splice.round.{IssuingMiningRound, OpenMiningRound} +import org.lfdecentralizedtrust.splice.codegen.java.splice.externalpartyamuletrules.{ + ExternalPartyAmuletRules, + TransferCommandCounter, +} +import org.lfdecentralizedtrust.splice.codegen.java.splice.round.{ + IssuingMiningRound, + OpenMiningRound, +} import org.lfdecentralizedtrust.splice.codegen.java.splice.types.Round 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, MigrationSchedule} +import org.lfdecentralizedtrust.splice.http.v0.definitions.{ + GetBulkObjectChecksumsResponse, + GetDsoInfoResponse, + GetRewardAccountingActivityTotalsResponse, + GetRewardAccountingBatchResponse, + GetRewardAccountingRootHashResponse, + HoldingsSummaryRequestV1, + HoldingsSummaryResponse, + HoldingsSummaryResponseV1, + LookupTransferCommandStatusResponse, + MigrationSchedule, +} import org.lfdecentralizedtrust.splice.scan.admin.api.client.ScanConnection.* import org.lfdecentralizedtrust.splice.scan.admin.api.client.commands.HttpScanAppClient import org.lfdecentralizedtrust.splice.scan.admin.api.client.commands.HttpScanAppClient.TransferContextWithInstances @@ -31,7 +51,10 @@ import com.digitalasset.canton.topology.{ParticipantId, PartyId, SynchronizerId} import com.digitalasset.canton.tracing.TraceContext import io.grpc.Status import org.apache.pekko.stream.Materializer -import org.lfdecentralizedtrust.splice.codegen.java.splice.dsorules.{DsoRules_CloseVoteRequestResult, VoteRequest} +import org.lfdecentralizedtrust.splice.codegen.java.splice.dsorules.{ + DsoRules_CloseVoteRequestResult, + VoteRequest, +} import org.lfdecentralizedtrust.splice.http.v0.definitions.HoldingsSummaryRequest.RecordTimeMatch import org.lfdecentralizedtrust.splice.metrics.ScanConnectionMetrics 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 1b4d2e1231..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 @@ -12,32 +12,92 @@ import com.daml.tls.TlsClientConfig import com.digitalasset.canton.config.RequireTypes import com.digitalasset.canton.config.RequireTypes.NonNegativeInt import org.lfdecentralizedtrust.splice.admin.api.client.commands.HttpCommand -import org.lfdecentralizedtrust.splice.codegen.java.splice.amulet.{FeaturedAppRight, UnclaimedDevelopmentFundCoupon} -import org.lfdecentralizedtrust.splice.codegen.java.splice.amuletrules.{AmuletRules, AppTransferContext, TransferPreapproval} -import org.lfdecentralizedtrust.splice.codegen.java.splice.externalpartyamuletrules.{ExternalPartyAmuletRules, TransferCommandCounter} -import org.lfdecentralizedtrust.splice.codegen.java.splice.round.{ClosedMiningRound, IssuingMiningRound, OpenMiningRound} +import org.lfdecentralizedtrust.splice.codegen.java.splice.amulet.{ + FeaturedAppRight, + UnclaimedDevelopmentFundCoupon, +} +import org.lfdecentralizedtrust.splice.codegen.java.splice.amuletrules.{ + AmuletRules, + AppTransferContext, + TransferPreapproval, +} +import org.lfdecentralizedtrust.splice.codegen.java.splice.externalpartyamuletrules.{ + ExternalPartyAmuletRules, + TransferCommandCounter, +} +import org.lfdecentralizedtrust.splice.codegen.java.splice.round.{ + ClosedMiningRound, + IssuingMiningRound, + OpenMiningRound, +} import org.lfdecentralizedtrust.splice.codegen.java.splice.ans as ansCodegen import org.lfdecentralizedtrust.splice.codegen.java.splice.ans.AnsRules import org.lfdecentralizedtrust.splice.config.SpliceInstanceNamesConfig import org.lfdecentralizedtrust.splice.http.v0.{definitions, scan as http} -import org.lfdecentralizedtrust.tokenstandard.{allocation, allocationinstruction, metadata, transferinstruction} -import org.lfdecentralizedtrust.splice.http.v0.scan.{ForceAcsSnapshotNowResponse, GetBulkObjectChecksumsResponse, GetDateOfFirstSnapshotAfterResponse, GetDateOfMostRecentSnapshotBeforeResponse, GetLsuResponse, ListBulkAcsSnapshotObjectsResponse, ListBulkUpdateHistoryObjectsResponse} -import org.lfdecentralizedtrust.splice.scan.admin.http.{CompactJsonScanHttpEncodings, ProtobufJsonScanHttpEncodings} +import org.lfdecentralizedtrust.tokenstandard.{ + allocation, + allocationinstruction, + metadata, + transferinstruction, +} +import org.lfdecentralizedtrust.splice.http.v0.scan.{ + ForceAcsSnapshotNowResponse, + GetBulkObjectChecksumsResponse, + GetDateOfFirstSnapshotAfterResponse, + GetDateOfMostRecentSnapshotBeforeResponse, + GetLsuResponse, + ListBulkAcsSnapshotObjectsResponse, + ListBulkUpdateHistoryObjectsResponse, +} +import org.lfdecentralizedtrust.splice.scan.admin.http.{ + CompactJsonScanHttpEncodings, + ProtobufJsonScanHttpEncodings, +} import org.lfdecentralizedtrust.splice.store.HistoryBackfilling.SourceMigrationInfo import org.lfdecentralizedtrust.splice.store.{MultiDomainAcsStore, VoteResultsFilters} import org.lfdecentralizedtrust.splice.store.UpdateHistory.UpdateHistoryResponse -import org.lfdecentralizedtrust.splice.util.{ChoiceContextWithDisclosures, Codec, Contract, ContractWithState, DomainRecordTimeRange, FactoryChoiceWithDisclosures, PackageQualifiedName, TemplateJsonDecoder} +import org.lfdecentralizedtrust.splice.util.{ + ChoiceContextWithDisclosures, + Codec, + Contract, + ContractWithState, + DomainRecordTimeRange, + FactoryChoiceWithDisclosures, + PackageQualifiedName, + TemplateJsonDecoder, +} import com.digitalasset.canton.data.CantonTimestamp import com.digitalasset.canton.logging.pretty.{Pretty, PrettyPrinting} import com.digitalasset.canton.synchronizer.sequencer.block.bftordering.bindings.p2p.grpc.P2PGrpcNetworking.P2PEndpoint import com.digitalasset.canton.synchronizer.sequencer.block.bftordering.core.BftBlockOrdererConfig.P2PEndpointConfig -import com.digitalasset.canton.topology.{Member, ParticipantId, PartyId, PhysicalSynchronizerId, SequencerId, SynchronizerId} +import com.digitalasset.canton.topology.{ + Member, + ParticipantId, + PartyId, + PhysicalSynchronizerId, + SequencerId, + SynchronizerId, +} import com.digitalasset.daml.lf.data.Time.Timestamp import com.google.protobuf.ByteString import org.apache.pekko.stream.scaladsl.Source -import org.lfdecentralizedtrust.splice.codegen.java.splice.api.token.{allocationinstructionv1, allocationinstructionv2, allocationv1, allocationv2, metadatav1, transferinstructionv1, transferinstructionv2} -import org.lfdecentralizedtrust.splice.codegen.java.splice.dsorules.{DsoRules_CloseVoteRequestResult, VoteRequest} -import org.lfdecentralizedtrust.splice.scan.admin.api.client.{BulkStorageDownloadResponse, ScanStreamClient} +import org.lfdecentralizedtrust.splice.codegen.java.splice.api.token.{ + allocationinstructionv1, + allocationinstructionv2, + allocationv1, + allocationv2, + metadatav1, + transferinstructionv1, + transferinstructionv2, +} +import org.lfdecentralizedtrust.splice.codegen.java.splice.dsorules.{ + DsoRules_CloseVoteRequestResult, + VoteRequest, +} +import org.lfdecentralizedtrust.splice.scan.admin.api.client.{ + BulkStorageDownloadResponse, + ScanStreamClient, +} import java.util.Base64 import java.time.Instant 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 9747ae98fb..b4a9535e50 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 @@ -16,7 +16,14 @@ import com.digitalasset.canton.time.Clock import com.digitalasset.canton.topology.store.TimeQuery import com.digitalasset.canton.topology.{Member, PartyId, SynchronizerId} import com.digitalasset.canton.tracing.TraceContext -import com.digitalasset.canton.util.{ByteStringUtil, ErrorUtil, GrpcStreamingUtils, MaxBytesToDecompress, MonadUtil, ResourceUtil} +import com.digitalasset.canton.util.{ + ByteStringUtil, + ErrorUtil, + GrpcStreamingUtils, + MaxBytesToDecompress, + MonadUtil, + ResourceUtil, +} import com.digitalasset.canton.util.ShowUtil.* import com.digitalasset.daml.lf.data.Time.Timestamp import com.github.blemale.scaffeine.{Cache, Scaffeine} @@ -29,28 +36,82 @@ import org.lfdecentralizedtrust.splice.codegen.java.splice.{amulet, ans as ansCo import org.lfdecentralizedtrust.splice.codegen.java.splice.amuletrules.AmuletRules import org.lfdecentralizedtrust.splice.codegen.java.splice.dso.decentralizedsynchronizer.SynchronizerNodeConfig import org.lfdecentralizedtrust.splice.codegen.java.splice.dso.svstate.SvNodeState -import org.lfdecentralizedtrust.splice.codegen.java.splice.externalpartyamuletrules.{ExternalPartyAmuletRules, TransferCommand} -import org.lfdecentralizedtrust.splice.codegen.java.splice.round.{ClosedMiningRound, IssuingMiningRound, OpenMiningRound, SummarizingMiningRound} +import org.lfdecentralizedtrust.splice.codegen.java.splice.externalpartyamuletrules.{ + ExternalPartyAmuletRules, + TransferCommand, +} +import org.lfdecentralizedtrust.splice.codegen.java.splice.round.{ + ClosedMiningRound, + IssuingMiningRound, + OpenMiningRound, + SummarizingMiningRound, +} import org.lfdecentralizedtrust.splice.config.{SpliceInstanceNamesConfig, Thresholds} -import org.lfdecentralizedtrust.splice.environment.{PackageVersionSupport, ParticipantAdminConnection, SequencerAdminConnection, SynchronizerNodeService} -import org.lfdecentralizedtrust.splice.environment.TopologyAdminConnection.{TopologySnapshot, TopologyTransactionType} +import org.lfdecentralizedtrust.splice.environment.{ + PackageVersionSupport, + ParticipantAdminConnection, + SequencerAdminConnection, + SynchronizerNodeService, +} +import org.lfdecentralizedtrust.splice.environment.TopologyAdminConnection.{ + TopologySnapshot, + TopologyTransactionType, +} import org.lfdecentralizedtrust.splice.environment.TopologyAdminConnection.TopologyTransactionType.AuthorizedState -import org.lfdecentralizedtrust.splice.http.{HttpFeatureSupportHandler, HttpValidatorLicensesHandler, HttpVotesHandler, UrlValidator} +import org.lfdecentralizedtrust.splice.http.{ + HttpFeatureSupportHandler, + HttpValidatorLicensesHandler, + HttpVotesHandler, + UrlValidator, +} import org.lfdecentralizedtrust.splice.http.v0.{definitions, scan as v0} -import org.lfdecentralizedtrust.splice.http.v0.definitions.{AcsRequest, BatchListVotesByVoteRequestsRequest, CountVoteResultsRequest, DamlValueEncoding, ErrorResponse, EventHistoryRequest, GetBulkObjectChecksumsRequest, HoldingsStateRequest, HoldingsSummaryRequest, HoldingsSummaryRequestV1, ListBulkUpdateHistoryObjectsRequest, ListVoteResultsRequest, MaybeCachedContractWithState, PreviousSvRewardWeightRequest, PreviousSvRewardWeightResponse, UpdateHistoryItem, UpdateHistoryItemV2WithHash, UpdateHistoryRequestV2, UpdateHistoryTransactionV2WithHash} +import org.lfdecentralizedtrust.splice.http.v0.definitions.{ + AcsRequest, + BatchListVotesByVoteRequestsRequest, + CountVoteResultsRequest, + DamlValueEncoding, + ErrorResponse, + EventHistoryRequest, + GetBulkObjectChecksumsRequest, + HoldingsStateRequest, + HoldingsSummaryRequest, + HoldingsSummaryRequestV1, + ListBulkUpdateHistoryObjectsRequest, + ListVoteResultsRequest, + MaybeCachedContractWithState, + PreviousSvRewardWeightRequest, + PreviousSvRewardWeightResponse, + UpdateHistoryItem, + UpdateHistoryItemV2WithHash, + UpdateHistoryRequestV2, + UpdateHistoryTransactionV2WithHash, +} import org.lfdecentralizedtrust.splice.http.v0.scan.ScanResource import org.lfdecentralizedtrust.splice.scan.ScanSynchronizerNode import org.lfdecentralizedtrust.splice.scan.admin.http.ScanHttpEncodings.updateV1ToUpdateV2 import org.lfdecentralizedtrust.splice.scan.config.{CantonBftPeerConfig, ScanRollForwardLsuConfig} import org.lfdecentralizedtrust.splice.scan.dso.DsoAnsResolver -import org.lfdecentralizedtrust.splice.scan.store.{AcsSnapshotStore, AppActivityStore, ScanEventStore, ScanStore, TxLogEntry} +import org.lfdecentralizedtrust.splice.scan.store.{ + AcsSnapshotStore, + AppActivityStore, + ScanEventStore, + ScanStore, + TxLogEntry, +} import org.lfdecentralizedtrust.splice.scan.store.bulk.BulkStorageReader import org.lfdecentralizedtrust.splice.scan.store.AcsSnapshotStore.QueryAcsSnapshotResult import org.lfdecentralizedtrust.splice.scan.store.bulk.AcsSnapshotBulkStorage.AcsSnapshotObjects import org.lfdecentralizedtrust.splice.scan.store.bulk.UpdateHistoryBulkStorage.UpdateHistoryObjectsResponse import org.lfdecentralizedtrust.splice.store.AppStoreWithIngestion.SpliceLedgerConnectionPriority import org.lfdecentralizedtrust.splice.store.MultiDomainAcsStore.TxLogBackfillingState -import org.lfdecentralizedtrust.splice.store.{AppStore, AppStoreWithIngestion, PageLimit, SortOrder, VoteResultsFilters, VotesStore} +import org.lfdecentralizedtrust.splice.store.{ + AppStore, + AppStoreWithIngestion, + PageLimit, + SortOrder, + VoteResultsFilters, + VotesStore, +} import org.lfdecentralizedtrust.splice.store.S3BucketConnection.ObjectKeyAndChecksum import org.lfdecentralizedtrust.splice.store.UpdateHistory.BackfillingState import org.lfdecentralizedtrust.splice.store.UpdateHistory @@ -58,7 +119,13 @@ import org.lfdecentralizedtrust.splice.store.UpdateHistory import java.lang.IllegalStateException import scala.collection.immutable.SortedMap import org.lfdecentralizedtrust.splice.scan.store.db.DbScanAppRewardsStore -import org.lfdecentralizedtrust.splice.util.{Codec, Contract, ContractWithState, PackageQualifiedName, QualifiedName} +import org.lfdecentralizedtrust.splice.util.{ + Codec, + Contract, + ContractWithState, + PackageQualifiedName, + QualifiedName, +} import org.lfdecentralizedtrust.splice.util.PrettyInstances.* import java.io.ByteArrayInputStream @@ -2583,7 +2650,9 @@ class HttpScanHandler( } } - override def getBulkObjectChecksums(respond: ScanResource.GetBulkObjectChecksumsResponse.type)(body: GetBulkObjectChecksumsRequest)(extracted: TraceContext): Future[ScanResource.GetBulkObjectChecksumsResponse] = { + override def getBulkObjectChecksums(respond: ScanResource.GetBulkObjectChecksumsResponse.type)( + body: GetBulkObjectChecksumsRequest + )(extracted: TraceContext): Future[ScanResource.GetBulkObjectChecksumsResponse] = { implicit val tc = extracted withSpan(s"$workflowId.getBulkObjectChecksums") { _ => _ => bulkStorage.fold( 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..9f14879b00 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.http.scaladsl.model.StatusCodes import org.apache.pekko.stream.scaladsl.{Flow, Sink, 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,65 @@ 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) + 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(_.nonEmpty) + logger.debug(s"Consensus achieved on ${consensusChecksums.length} objects") + val consensus = bftChecksums.checksums.filter(_.nonEmpty) == objects.map(_.checksum) + if (consensusChecksums.length == objects.length && !consensus) { + logger.warn( + s"BFT consensus checksums do not match the expected checksums for all objects. Expected: ${objects + .map(_.checksum) + .mkString(", ")}, got: ${consensusChecksums.mkString(", ")}" + ) + } + consensus + case None => + false + } + } } // TODO(#5884): implement the BFT check @@ -136,17 +191,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/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 index ad6f5550c6..bed8bcfc0e 100644 --- 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 @@ -1,3 +1,6 @@ +// 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 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..1c5fe5cfb5 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, // TODO: enable here or in a different test ) 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..592980dc2d 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 @@ -11,10 +11,12 @@ import com.digitalasset.canton.{HasActorSystem, HasExecutionContext} import org.slf4j.event.Level import org.apache.pekko.stream.scaladsl.Keep import org.apache.pekko.stream.testkit.scaladsl.{TestSink, TestSource} +import org.lfdecentralizedtrust.splice.http.HttpClient import org.lfdecentralizedtrust.splice.scan.config.BulkStorageConfig 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 @@ -29,7 +31,9 @@ class BulkStorageCommitFromStagingTest with SplicePostgresTest { override val initialBuckets = Seq("staging", "committed") - val appConfig = BulkStorageConfig() + val appConfig = BulkStorageConfig( + bftCheckEnabled = false // TODO: enable here or in a different test + ) "BulkStorageCommitFromStaging" should { "successfully move objects from staging to committed S3 bucket" in { @@ -68,11 +72,21 @@ class BulkStorageCommitFromStagingTest committedS3Connection: S3BucketConnectionForUnitTests, objsWithDigests: Seq[ObjectKeyAndChecksum], ) = { + 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 flow = 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, ) 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..f0daba0c31 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, // TODO: enable here or in a different test ) "UpdateHistoryBulkStorage" should { From bdefa9fc92006fe58cfb4fc0976544773cef5603 Mon Sep 17 00:00:00 2001 From: Itai Segall Date: Fri, 17 Jul 2026 16:23:48 +0000 Subject: [PATCH 07/17] [ci] actually support the disable flag Signed-off-by: Itai Segall --- .../tests/ScanTimeBasedIntegrationTest.scala | 2 + .../bulk/BulkStorageCommitFromStaging.scala | 75 ++++++++++--------- 2 files changed, 42 insertions(+), 35 deletions(-) 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..5b1f82525b 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, // FIXME: this is only one SV, so in theory should work with bft enabled? It does not, investigate that. ), publicUrl = Some(Uri("http://foo.bar.com")), ) 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 9f14879b00..bfd3d60b49 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 @@ -52,42 +52,47 @@ class BulkStorageCommitFromStaging[T]( logger.debug( s"Checking BFT agreement for objects: ${objects.map(_.key).mkString(", ")}" ) - 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(_.nonEmpty) - logger.debug(s"Consensus achieved on ${consensusChecksums.length} objects") - val consensus = bftChecksums.checksums.filter(_.nonEmpty) == objects.map(_.checksum) - if (consensusChecksums.length == objects.length && !consensus) { - logger.warn( - s"BFT consensus checksums do not match the expected checksums for all objects. Expected: ${objects - .map(_.checksum) - .mkString(", ")}, got: ${consensusChecksums.mkString(", ")}" - ) - } - consensus - case None => - false + 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(_.nonEmpty) + logger.debug(s"Consensus achieved on ${consensusChecksums.length} objects") + val consensus = bftChecksums.checksums.filter(_.nonEmpty) == objects.map(_.checksum) + if (consensusChecksums.length == objects.length && !consensus) { + logger.warn( + s"BFT consensus checksums do not match the expected checksums for all objects. 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 From 83825c2ef8543218d0dc3031b0c889420ba5867c Mon Sep 17 00:00:00 2001 From: Itai Segall Date: Fri, 17 Jul 2026 17:23:57 +0000 Subject: [PATCH 08/17] [ci] rate limits Signed-off-by: Itai Segall --- cluster/configs/shared/rate-limits/unlimited.yaml | 4 ++++ cluster/deployment/scratchneta/config.resolved.yaml | 3 +++ cluster/deployment/scratchnetb/config.resolved.yaml | 3 +++ cluster/deployment/scratchnetc/config.resolved.yaml | 3 +++ cluster/deployment/scratchnetd/config.resolved.yaml | 3 +++ cluster/deployment/scratchnete/config.resolved.yaml | 3 +++ cluster/expected/canton-network/expected.json | 8 ++++++++ 7 files changed, 27 insertions(+) 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" From f449967eaeee81511e1143922825263c18fe6c38 Mon Sep 17 00:00:00 2001 From: Itai Segall Date: Fri, 17 Jul 2026 18:01:15 +0000 Subject: [PATCH 09/17] [ci] another expected Signed-off-by: Itai Segall --- cluster/expected/sv-runbook/expected.json | 4 ++++ 1 file changed, 4 insertions(+) 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" From c123e34508d0a4c8c253f549d94100a03408dd0c Mon Sep 17 00:00:00 2001 From: Itai Segall Date: Fri, 17 Jul 2026 20:43:35 +0000 Subject: [PATCH 10/17] [ci] basic unit test Signed-off-by: Itai Segall --- .../BulkStorageCommitFromStagingTest.scala | 141 ++++++++++++++---- 1 file changed, 111 insertions(+), 30 deletions(-) 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 592980dc2d..47fa6ccc27 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 @@ -4,15 +4,23 @@ package org.lfdecentralizedtrust.splice.scan.store.bulk 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.tracing.TraceContext import com.digitalasset.canton.{HasActorSystem, HasExecutionContext} +import org.apache.pekko.NotUsed +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 import org.lfdecentralizedtrust.splice.scan.config.BulkStorageConfig +import org.lfdecentralizedtrust.splice.scan.store.ScanStore import org.lfdecentralizedtrust.splice.store.S3BucketConnection.ObjectKeyAndChecksum import org.lfdecentralizedtrust.splice.store.{HasS3Mock, StoreTestBase} import org.lfdecentralizedtrust.splice.store.db.SplicePostgresTest @@ -20,7 +28,7 @@ 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.* class BulkStorageCommitFromStagingTest @@ -31,16 +39,21 @@ class BulkStorageCommitFromStagingTest with SplicePostgresTest { override val initialBuckets = Seq("staging", "committed") - val appConfig = BulkStorageConfig( - bftCheckEnabled = false // TODO: enable here or in a different test - ) + 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) } @@ -58,38 +71,106 @@ 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], - ) = { - 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 flow = 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, - ) + "BulkStorageCommitFromStaging with BFT reads enabled" should { + val appConfig = BulkStorageConfig() + "successfully move objects from staging to committed S3 bucket when there's full consensus" in { + val (stagingS3Connection, committedS3Connection, objsWithDigests) = setupTest + + val mockConnection = mock[BftScanConnection] + when( + mockConnection + .getBulkObjectChecksums(any[Seq[String]])(any[ExecutionContext], any[TraceContext]) + ) + .thenReturn( + Future.successful( + new GetBulkObjectChecksumsResponse(objsWithDigests.map(_.checksum).toVector) + ) + ) + + triggerCopyFlowAndAssertCompletion( + newCopyFlow(mockConnection, stagingS3Connection, committedS3Connection, objsWithDigests) + ) + + assertObjectsMoved(stagingS3Connection, committedS3Connection, objsWithDigests) + } + + def newCopyFlow( + bftScanConnection: BftScanConnection, + stagingS3Connection: S3BucketConnectionForUnitTests, + committedS3Connection: S3BucketConnectionForUnitTests, + objsWithDigests: Seq[ObjectKeyAndChecksum], + ) = { + new BulkStorageCommitFromStaging[String]( + stagingS3Connection, + committedS3Connection, + _ => Future.successful(objsWithDigests), + appConfig, + null, // we're mocking the bft reads + null, // we're mocking the bft reads + null, // we're mocking the bft reads + null, // we're mocking the bft reads + null, // we're mocking the bft reads + null, // we're mocking the bft reads + null, // we're mocking the bft reads + 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(bftScanConnection) + }.getFlow + } + } + + private def triggerCopyFlowAndAssertCompletion( + flow: Flow[String, String, NotUsed] + ) = { val (pub, sub) = TestSource .probe[String] .via(flow) From 148a1e3239638cc8649657c9bf2ec933251adf6f Mon Sep 17 00:00:00 2001 From: Itai Segall Date: Fri, 17 Jul 2026 21:28:05 +0000 Subject: [PATCH 11/17] [ci] more testing Signed-off-by: Itai Segall --- .../BulkStorageCommitFromStagingTest.scala | 77 ++++++++++++++++++- 1 file changed, 76 insertions(+), 1 deletion(-) 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 47fa6ccc27..92d9e17c28 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,6 +3,7 @@ package org.lfdecentralizedtrust.splice.scan.store.bulk +import com.digitalasset.canton.config.NonNegativeFiniteDuration import com.digitalasset.canton.lifecycle.FutureUnlessShutdown import com.digitalasset.canton.logging.{NamedLoggerFactory, SuppressionRule} import com.digitalasset.canton.resource.DbStorage @@ -10,10 +11,12 @@ import com.digitalasset.canton.time.Clock import com.digitalasset.canton.tracing.TraceContext import com.digitalasset.canton.{HasActorSystem, HasExecutionContext} import org.apache.pekko.NotUsed +import org.apache.pekko.http.scaladsl.model.StatusCodes import org.apache.pekko.stream.Materializer import org.slf4j.event.Level import org.apache.pekko.stream.scaladsl.{Flow, Keep} import org.apache.pekko.stream.testkit.scaladsl.{TestSink, TestSource} +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 @@ -30,6 +33,7 @@ import java.security.MessageDigest import java.util.Base64 import scala.concurrent.{ExecutionContext, ExecutionContextExecutor, Future} import scala.jdk.CollectionConverters.* +import scala.concurrent.duration.* class BulkStorageCommitFromStagingTest extends StoreTestBase @@ -105,7 +109,9 @@ class BulkStorageCommitFromStagingTest } "BulkStorageCommitFromStaging with BFT reads enabled" should { - val appConfig = BulkStorageConfig() + 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 @@ -128,6 +134,75 @@ class BulkStorageCommitFromStagingTest assertObjectsMoved(stagingS3Connection, committedS3Connection, objsWithDigests) } + "wait until all objects are known to the peers" in { + val (stagingS3Connection, committedS3Connection, objsWithDigests) = setupTest + + val mockConnection = mock[BftScanConnection] + + val flow = + newCopyFlow(mockConnection, stagingS3Connection, committedS3Connection, objsWithDigests) + + val (pub, sub) = TestSource + .probe[String] + .via(flow) + .toMat(TestSink.probe[String])(Keep.both) + .run() + + clue("When one object is not known to the peers, the copy flow should not complete") { + when( + mockConnection + .getBulkObjectChecksums(any[Seq[String]])(any[ExecutionContext], any[TraceContext]) + ) + .thenReturn( + Future.successful( + new GetBulkObjectChecksumsResponse( + objsWithDigests.map(_.checksum).toVector.updated(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 + } + + clue("Simulate disagreement across peers") { + when( + mockConnection + .getBulkObjectChecksums(any[Seq[String]])(any[ExecutionContext], any[TraceContext]) + ) + .thenReturn( + Future.failed( + HttpErrorWithHttpCode(StatusCodes.BadGateway, "Simulated disagreement across peers") + ) + ) + sub.expectNoMessage(20.seconds) + } + + clue( + "Make the missing object known to all peers, the copy flow should now complete successfully" + ) { + when( + mockConnection + .getBulkObjectChecksums(any[Seq[String]])(any[ExecutionContext], any[TraceContext]) + ) + .thenReturn( + Future.successful( + new GetBulkObjectChecksumsResponse( + objsWithDigests.map(_.checksum).toVector + ) + ) + ) + + sub.expectNext(20.seconds, "go") + assertObjectsMoved(stagingS3Connection, committedS3Connection, objsWithDigests) + } + } + def newCopyFlow( bftScanConnection: BftScanConnection, stagingS3Connection: S3BucketConnectionForUnitTests, From ba5201b2fdb924e700436ffb34d748de7c350568 Mon Sep 17 00:00:00 2001 From: Itai Segall Date: Mon, 20 Jul 2026 22:15:22 +0000 Subject: [PATCH 12/17] [ci] test works Signed-off-by: Itai Segall --- .../api/client/SingleScanConnection.scala | 2 +- .../bulk/BulkStorageCommitFromStaging.scala | 2 +- .../BulkStorageCommitFromStagingTest.scala | 222 +++++++++++++----- 3 files changed, 166 insertions(+), 60 deletions(-) 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 86bdb4de4c..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 @@ -97,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, 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 bfd3d60b49..6e244b6b75 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 @@ -79,7 +79,7 @@ class BulkStorageCommitFromStaging[T]( logger.debug(s"Consensus achieved on ${consensusChecksums.length} objects") val consensus = bftChecksums.checksums.filter(_.nonEmpty) == objects.map(_.checksum) if (consensusChecksums.length == objects.length && !consensus) { - logger.warn( + logger.error( s"BFT consensus checksums do not match the expected checksums for all objects. Expected: ${objects .map(_.checksum) .mkString(", ")}, got: ${consensusChecksums.mkString(", ")}" 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 92d9e17c28..e8ac419fd6 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 @@ -8,22 +8,28 @@ import com.digitalasset.canton.lifecycle.FutureUnlessShutdown 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.StatusCodes +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.{Flow, Keep} import org.apache.pekko.stream.testkit.scaladsl.{TestSink, TestSource} -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.http.v0.definitions.GetBulkObjectChecksumsResponse -import org.lfdecentralizedtrust.splice.scan.admin.api.client.BftScanConnection +import org.lfdecentralizedtrust.splice.scan.admin.api.client.{ + BftScanConnection, + SingleScanConnection, +} import org.lfdecentralizedtrust.splice.scan.config.BulkStorageConfig -import org.lfdecentralizedtrust.splice.scan.store.ScanStore +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 @@ -40,7 +46,8 @@ class BulkStorageCommitFromStagingTest with HasExecutionContext with HasActorSystem with HasS3Mock - with SplicePostgresTest { + with SplicePostgresTest + with HasRetryProvider { override val initialBuckets = Seq("staging", "committed") implicit val httpClient: HttpClient = null @@ -116,31 +123,34 @@ class BulkStorageCommitFromStagingTest "successfully move objects from staging to committed S3 bucket when there's full consensus" in { val (stagingS3Connection, committedS3Connection, objsWithDigests) = setupTest - val mockConnection = mock[BftScanConnection] - when( - mockConnection - .getBulkObjectChecksums(any[Seq[String]])(any[ExecutionContext], any[TraceContext]) - ) - .thenReturn( - Future.successful( - new GetBulkObjectChecksumsResponse(objsWithDigests.map(_.checksum).toVector) - ) - ) + val mockScanConnections = new MockScanConnections(objsWithDigests) + Seq.range(0, 7).foreach { i => + mockScanConnections.scanAgrees(i) + } - triggerCopyFlowAndAssertCompletion( - newCopyFlow(mockConnection, stagingS3Connection, committedS3Connection, objsWithDigests) + val flow = newCopyFlow( + stagingS3Connection, + committedS3Connection, + objsWithDigests, + mockScanConnections, ) + triggerCopyFlowAndAssertCompletion(flow) + assertObjectsMoved(stagingS3Connection, committedS3Connection, objsWithDigests) } "wait until all objects are known to the peers" in { val (stagingS3Connection, committedS3Connection, objsWithDigests) = setupTest - val mockConnection = mock[BftScanConnection] + val mockScanConnections = new MockScanConnections(objsWithDigests) - val flow = - newCopyFlow(mockConnection, stagingS3Connection, committedS3Connection, objsWithDigests) + val flow = newCopyFlow( + stagingS3Connection, + committedS3Connection, + objsWithDigests, + mockScanConnections, + ) val (pub, sub) = TestSource .probe[String] @@ -148,79 +158,166 @@ class BulkStorageCommitFromStagingTest .toMat(TestSink.probe[String])(Keep.both) .run() - clue("When one object is not known to the peers, the copy flow should not complete") { + 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 + } + + 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)) + loggerFactory.assertLogsSeq(SuppressionRule.LevelAndAbove(Level.ERROR))( + { + sub.expectNoMessage(20.seconds) + }, + logEntries => + forAtLeast(1, logEntries)( + _.message should include( + "BFT consensus checksums do not match the expected checksums for all objects" + ) + ), + ) + 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) + } + } catch { + case ex: Throwable => + pub.sendComplete() + 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( - mockConnection + singleScanConnections(idx) .getBulkObjectChecksums(any[Seq[String]])(any[ExecutionContext], any[TraceContext]) ) .thenReturn( Future.successful( - new GetBulkObjectChecksumsResponse( - objsWithDigests.map(_.checksum).toVector.updated(1, "") - ) + new GetBulkObjectChecksumsResponse(objsWithDigests.map(_.checksum).toVector) ) ) - 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 + () } - clue("Simulate disagreement across peers") { + def scanDisagreesOnDigest(scanIdx: Integer, objIdx: Integer): Unit = { when( - mockConnection + singleScanConnections(scanIdx) .getBulkObjectChecksums(any[Seq[String]])(any[ExecutionContext], any[TraceContext]) ) .thenReturn( - Future.failed( - HttpErrorWithHttpCode(StatusCodes.BadGateway, "Simulated disagreement across peers") + Future.successful( + new GetBulkObjectChecksumsResponse( + objsWithDigests.map(_.checksum).updated(objIdx, "wrong-digest").toVector + ) ) ) - sub.expectNoMessage(20.seconds) } - clue( - "Make the missing object known to all peers, the copy flow should now complete successfully" - ) { + def scanMissingAnObject(scanIdx: Integer, objIdx: Integer): Unit = { when( - mockConnection + singleScanConnections(scanIdx) .getBulkObjectChecksums(any[Seq[String]])(any[ExecutionContext], any[TraceContext]) ) .thenReturn( Future.successful( new GetBulkObjectChecksumsResponse( - objsWithDigests.map(_.checksum).toVector + objsWithDigests.map(_.checksum).updated(objIdx, "").toVector ) ) ) - - sub.expectNext(20.seconds, "go") - assertObjectsMoved(stagingS3Connection, committedS3Connection, objsWithDigests) } + + 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( - bftScanConnection: BftScanConnection, stagingS3Connection: S3BucketConnectionForUnitTests, committedS3Connection: S3BucketConnectionForUnitTests, objsWithDigests: Seq[ObjectKeyAndChecksum], + mockScanConnections: MockScanConnections, ) = { new BulkStorageCommitFromStaging[String]( stagingS3Connection, committedS3Connection, _ => Future.successful(objsWithDigests), appConfig, - null, // we're mocking the bft reads - null, // we're mocking the bft reads - null, // we're mocking the bft reads - null, // we're mocking the bft reads - null, // we're mocking the bft reads - null, // we're mocking the bft reads - null, // we're mocking the bft reads + 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( @@ -238,7 +335,7 @@ class BulkStorageCommitFromStagingTest mat: Materializer, httpClient: HttpClient, templateJsonDecoder: TemplateJsonDecoder, - ): Future[BftScanConnection] = Future.successful(bftScanConnection) + ): Future[BftScanConnection] = Future.successful(mockScanConnections.bftConnection) }.getFlow } } @@ -252,9 +349,18 @@ class BulkStorageCommitFromStagingTest .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.sendComplete() + sub.cancel() + throw ex + } } private def assertObjectsMoved( From 06e1cc292f6455db98abdbe8030979f55c274d94 Mon Sep 17 00:00:00 2001 From: Itai Segall Date: Tue, 21 Jul 2026 16:06:45 +0000 Subject: [PATCH 13/17] [ci] optional instead of empty string Signed-off-by: Itai Segall --- apps/scan/src/main/openapi/scan.yaml | 5 ++++- .../splice/scan/admin/http/HttpScanHandler.scala | 2 +- .../scan/store/bulk/BulkStorageCommitFromStaging.scala | 7 ++++--- .../splice/scan/store/bulk/BulkStorageReader.scala | 7 ++----- 4 files changed, 11 insertions(+), 10 deletions(-) diff --git a/apps/scan/src/main/openapi/scan.yaml b/apps/scan/src/main/openapi/scan.yaml index 7f34ce4885..78c982d486 100644 --- a/apps/scan/src/main/openapi/scan.yaml +++ b/apps/scan/src/main/openapi/scan.yaml @@ -4260,7 +4260,10 @@ components: The list of checksums for the requested bulk storage objects (in the same order as the object_keys). type: array items: - type: string + type: object + properties: + value: + type: string BulkStorageObjectRef: type: object 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 b4a9535e50..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 @@ -2665,7 +2665,7 @@ class HttpScanHandler( bulkStorage.getObjectChecksums(body.objectKeys).map { checksums => ScanResource.GetBulkObjectChecksumsResponse.OK( definitions.GetBulkObjectChecksumsResponse( - checksums.toVector + checksums.map(definitions.GetBulkObjectChecksumsResponse.Checksums(_)).toVector ) ) } 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 6e244b6b75..8147e9b85b 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 @@ -75,12 +75,13 @@ class BulkStorageCommitFromStaging[T]( } yield { bft match { case Some(bftChecksums) => - val consensusChecksums = bftChecksums.checksums.filter(_.nonEmpty) + val consensusChecksums = bftChecksums.checksums.filter(_.value.isDefined) logger.debug(s"Consensus achieved on ${consensusChecksums.length} objects") - val consensus = bftChecksums.checksums.filter(_.nonEmpty) == objects.map(_.checksum) + val consensus = + bftChecksums.checksums.filter(_.value.isDefined) == objects.map(_.checksum) if (consensusChecksums.length == objects.length && !consensus) { logger.error( - s"BFT consensus checksums do not match the expected checksums for all objects. Expected: ${objects + 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(", ")}" ) 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 9ca07b5015..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 @@ -145,16 +145,13 @@ class BulkStorageReader( def getObjectChecksums( objectKeys: Seq[String] - ): Future[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)) match { - case Some(obj) => obj.checksum - case None => "" - } + committed.find(_.key == key).orElse(staging.find(_.key == key)).map(_.checksum) } } } From dd1d2ad302336fb94de69fb593b9b212220d775c Mon Sep 17 00:00:00 2001 From: Itai Segall Date: Tue, 21 Jul 2026 19:56:38 +0000 Subject: [PATCH 14/17] [ci] fix and cleanup Signed-off-by: Itai Segall --- .../bulk/BulkStorageCommitFromStaging.scala | 11 +++++---- .../BulkStorageCommitFromStagingTest.scala | 24 +++++++++++++++---- 2 files changed, 25 insertions(+), 10 deletions(-) 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 8147e9b85b..f338d3d112 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 @@ -9,7 +9,7 @@ import com.digitalasset.canton.tracing.TraceContext import org.apache.pekko.NotUsed import org.apache.pekko.actor.ActorSystem import org.apache.pekko.http.scaladsl.model.StatusCodes -import org.apache.pekko.stream.scaladsl.{Flow, Sink, Source} +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} @@ -78,7 +78,9 @@ class BulkStorageCommitFromStaging[T]( val consensusChecksums = bftChecksums.checksums.filter(_.value.isDefined) logger.debug(s"Consensus achieved on ${consensusChecksums.length} objects") val consensus = - bftChecksums.checksums.filter(_.value.isDefined) == objects.map(_.checksum) + 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 @@ -103,7 +105,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))) @@ -121,8 +123,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) } } } 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 e8ac419fd6..a4c48251e4 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 @@ -184,7 +184,7 @@ class BulkStorageCommitFromStagingTest logEntries => forAtLeast(1, logEntries)( _.message should include( - "BFT consensus checksums do not match the expected checksums for all objects" + "All objects are known to the BFT peers, but the checksums do not match" ) ), ) @@ -202,7 +202,7 @@ class BulkStorageCommitFromStagingTest } } catch { case ex: Throwable => - pub.sendComplete() + pub.sendError(ex) sub.cancel() throw ex } @@ -230,7 +230,12 @@ class BulkStorageCommitFromStagingTest ) .thenReturn( Future.successful( - new GetBulkObjectChecksumsResponse(objsWithDigests.map(_.checksum).toVector) + new GetBulkObjectChecksumsResponse( + objsWithDigests + .map(_.checksum) + .map(digest => new GetBulkObjectChecksumsResponse.Checksums(Some(digest))) + .toVector + ) ) ) () @@ -244,7 +249,11 @@ class BulkStorageCommitFromStagingTest .thenReturn( Future.successful( new GetBulkObjectChecksumsResponse( - objsWithDigests.map(_.checksum).updated(objIdx, "wrong-digest").toVector + objsWithDigests + .map(_.checksum) + .updated(objIdx, "wrong-digest") + .map(digest => new GetBulkObjectChecksumsResponse.Checksums(Some(digest))) + .toVector ) ) ) @@ -258,7 +267,12 @@ class BulkStorageCommitFromStagingTest .thenReturn( Future.successful( new GetBulkObjectChecksumsResponse( - objsWithDigests.map(_.checksum).updated(objIdx, "").toVector + objsWithDigests + .map(_.checksum) + .map(Some(_)) + .updated(objIdx, None) + .map(oDigest => new GetBulkObjectChecksumsResponse.Checksums(oDigest)) + .toVector ) ) ) From 5e4ca529b7edc46fdb75f906f491551a01d58765 Mon Sep 17 00:00:00 2001 From: Itai Segall Date: Tue, 21 Jul 2026 20:06:16 +0000 Subject: [PATCH 15/17] [ci] cleanups Signed-off-by: Itai Segall --- .../tests/ScanTimeBasedIntegrationTest.scala | 4 +- ...shotBulkStorageCommitFromStagingTest.scala | 2 +- .../BulkStorageCommitFromStagingTest.scala | 54 ++++++++++--------- .../bulk/UpdateHistoryBulkStorageTest.scala | 2 +- 4 files changed, 33 insertions(+), 29 deletions(-) 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 5b1f82525b..f0520bad4b 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,8 +71,8 @@ class ScanTimeBasedIntegrationTest updatesPollingInterval = NonNegativeFiniteDuration.ofSeconds(5), staging = Some(s3ConfigMock("staging")), committed = Some(s3ConfigMock("committed")), - bftCheckEnabled = - false, // FIXME: this is only one SV, so in theory should work with bft enabled? It does not, investigate that. + // We leave bft enabled here, but since we have only one scan node, that's not very informative. + // BFT checks are tested in a unit test, and are thus not covered here. ), publicUrl = Some(Uri("http://foo.bar.com")), ) 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 1c5fe5cfb5..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 @@ -55,7 +55,7 @@ class AcsSnapshotBulkStorageCommitFromStagingTest ) val appConfig = BulkStorageConfig( snapshotPollingInterval = NonNegativeFiniteDuration.ofSeconds(5), - bftCheckEnabled = false, // TODO: enable here or in a different test + bftCheckEnabled = false, // bft checks are tested elsewhere ) override val initialBuckets: Seq[String] = Seq("staging", "committed") 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 a4c48251e4..53e5b6496e 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 @@ -173,33 +173,37 @@ class BulkStorageCommitFromStagingTest committedS3Connection.listObjects.futureValue.contents().asScala shouldBe empty } - 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)) - loggerFactory.assertLogsSeq(SuppressionRule.LevelAndAbove(Level.ERROR))( - { + // 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) - }, - logEntries => - forAtLeast(1, logEntries)( - _.message should include( - "All objects are known to the BFT peers, but the checksums do not match" - ) - ), - ) - stagingS3Connection.listObjects.futureValue - .contents() - .asScala should have size objsWithDigests.size.toLong - committedS3Connection.listObjects.futureValue.contents().asScala shouldBe empty - - } + 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" + ) + ), + ) - 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) - } } catch { case ex: Throwable => pub.sendError(ex) 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 f0daba0c31..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 @@ -58,7 +58,7 @@ class UpdateHistoryBulkStorageTest ) val appConfig = BulkStorageConfig( updatesPollingInterval = NonNegativeFiniteDuration.ofSeconds(5), - bftCheckEnabled = false, // TODO: enable here or in a different test + bftCheckEnabled = false, // bft checks are tested elsewhere ) "UpdateHistoryBulkStorage" should { From 6181e8d852d7527ad135aeb60dd0d6e3aec88d10 Mon Sep 17 00:00:00 2001 From: Itai Segall Date: Tue, 21 Jul 2026 20:59:29 +0000 Subject: [PATCH 16/17] [ci] disable bft in integration test Signed-off-by: Itai Segall --- .../integration/tests/ScanTimeBasedIntegrationTest.scala | 3 +-- .../splice/scan/store/bulk/BulkStorageCommitFromStaging.scala | 2 +- .../scan/store/bulk/BulkStorageCommitFromStagingTest.scala | 4 ++-- 3 files changed, 4 insertions(+), 5 deletions(-) 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 f0520bad4b..29ddda7e9d 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,8 +71,7 @@ class ScanTimeBasedIntegrationTest updatesPollingInterval = NonNegativeFiniteDuration.ofSeconds(5), staging = Some(s3ConfigMock("staging")), committed = Some(s3ConfigMock("committed")), - // We leave bft enabled here, but since we have only one scan node, that's not very informative. - // BFT checks are tested in a unit test, and are thus not covered here. + 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/scala/org/lfdecentralizedtrust/splice/scan/store/bulk/BulkStorageCommitFromStaging.scala b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/bulk/BulkStorageCommitFromStaging.scala index f338d3d112..df58e8b8ec 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 @@ -76,7 +76,7 @@ class BulkStorageCommitFromStaging[T]( bft match { case Some(bftChecksums) => val consensusChecksums = bftChecksums.checksums.filter(_.value.isDefined) - logger.debug(s"Consensus achieved on ${consensusChecksums.length} objects") + 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) 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 53e5b6496e..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 @@ -140,7 +140,7 @@ class BulkStorageCommitFromStagingTest assertObjectsMoved(stagingS3Connection, committedS3Connection, objsWithDigests) } - "wait until all objects are known to the peers" in { + "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) @@ -375,7 +375,7 @@ class BulkStorageCommitFromStagingTest sub.expectComplete() } catch { case ex: Throwable => - pub.sendComplete() + pub.sendError(ex) sub.cancel() throw ex } From 335929e2b89ac08d86d612f3926d81448348a77b Mon Sep 17 00:00:00 2001 From: Itai Segall Date: Tue, 21 Jul 2026 21:11:55 +0000 Subject: [PATCH 17/17] [ci] fmt Signed-off-by: Itai Segall --- .../integration/tests/ScanTimeBasedIntegrationTest.scala | 3 ++- .../splice/scan/store/bulk/BulkStorageCommitFromStaging.scala | 4 +++- 2 files changed, 5 insertions(+), 2 deletions(-) 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 29ddda7e9d..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,7 +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. + 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/scala/org/lfdecentralizedtrust/splice/scan/store/bulk/BulkStorageCommitFromStaging.scala b/apps/scan/src/main/scala/org/lfdecentralizedtrust/splice/scan/store/bulk/BulkStorageCommitFromStaging.scala index df58e8b8ec..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 @@ -76,7 +76,9 @@ class BulkStorageCommitFromStaging[T]( 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") + 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)