diff --git a/apps/common/src/main/resources/db/migration/canton-network/postgres/stable/V073__sync_operator_acs_store.sql b/apps/common/src/main/resources/db/migration/canton-network/postgres/stable/V073__sync_operator_acs_store.sql new file mode 100644 index 0000000000..e6035df96f --- /dev/null +++ b/apps/common/src/main/resources/db/migration/canton-network/postgres/stable/V073__sync_operator_acs_store.sql @@ -0,0 +1,25 @@ +-- ACS store of the sync operator app, which only ingests MemberTraffic. +create table sync_operator_acs_store( + like acs_store_template including all, + + -- reestablish foreign key constraint as that one is not copied by the LIKE statement above + foreign key (store_id) references store_descriptors(id), + + -- index columns + ---------------- + + -- the member id in a MemberTraffic + member_traffic_member text, + + -- the synchronizer id in a MemberTraffic. Constant for this store, which is scoped to a single + -- synchronizer, but kept so the query and index match the SV and scan stores. + member_traffic_domain text, + + -- the purchased traffic in a MemberTraffic + total_traffic_purchased bigint +); + +create index sync_operator_acs_store_sid_mid_pn_tid_mtm_mtd + on sync_operator_acs_store (store_id, migration_id, package_name, template_id_qualified_name, + member_traffic_member, member_traffic_domain) + where member_traffic_member is not null; diff --git a/apps/common/src/main/scala/org/lfdecentralizedtrust/splice/store/db/MemberTrafficQueries.scala b/apps/common/src/main/scala/org/lfdecentralizedtrust/splice/store/db/MemberTrafficQueries.scala new file mode 100644 index 0000000000..23dd4de7b7 --- /dev/null +++ b/apps/common/src/main/scala/org/lfdecentralizedtrust/splice/store/db/MemberTrafficQueries.scala @@ -0,0 +1,51 @@ +// Copyright (c) 2024 Digital Asset (Switzerland) GmbH and/or its affiliates. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +package org.lfdecentralizedtrust.splice.store.db + +import com.digitalasset.canton.lifecycle.CloseContext +import com.digitalasset.canton.logging.NamedLogging +import com.digitalasset.canton.resource.DbStorage +import com.digitalasset.canton.topology.{Member, SynchronizerId} +import com.digitalasset.canton.tracing.TraceContext +import org.lfdecentralizedtrust.splice.codegen.java.splice.decentralizedsynchronizer.MemberTraffic +import org.lfdecentralizedtrust.splice.store.LimitHelpers +import org.lfdecentralizedtrust.splice.store.db.AcsQueries.AcsStoreId +import org.lfdecentralizedtrust.splice.util.QualifiedName +import org.lfdecentralizedtrust.splice.util.FutureUnlessShutdownUtil.futureUnlessShutdownToFuture +import slick.jdbc.canton.ActionBasedSQLInterpolation.Implicits.actionBasedSQLInterpolationCanton + +import scala.concurrent.{ExecutionContext, Future} + +/** Shared `MemberTraffic` queries for stores that ingest it with the standard index columns. */ +trait MemberTrafficQueries extends AcsJdbcTypes with LimitHelpers { this: NamedLogging => + + protected def sumPurchasedMemberTraffic( + storage: DbStorage, + acsTableName: String, + acsStoreId: AcsStoreId, + migrationId: Long, + memberId: Member, + synchronizerId: SynchronizerId, + )(implicit ec: ExecutionContext, tc: TraceContext, closeContext: CloseContext): Future[Long] = + for { + sum <- storage + .querySingle( + sql""" + select sum(total_traffic_purchased) + from #$acsTableName + where store_id = $acsStoreId + and migration_id = $migrationId + and package_name = ${MemberTraffic.PACKAGE_NAME} + and template_id_qualified_name = ${QualifiedName( + MemberTraffic.TEMPLATE_ID_WITH_PACKAGE_ID + )} + and member_traffic_member = ${lengthLimited(memberId.toProtoPrimitive)} + and member_traffic_domain = $synchronizerId + """.as[Long].headOption, + // the callers' method name, which is what the DB retry logs report + "getTotalPurchasedMemberTraffic", + ) + .value + } yield sum.getOrElse(0L) +} diff --git a/apps/common/src/main/scala/org/lfdecentralizedtrust/splice/store/db/SpliceDbLockCounters.scala b/apps/common/src/main/scala/org/lfdecentralizedtrust/splice/store/db/SpliceDbLockCounters.scala index 6143a0ca57..36492f957b 100644 --- a/apps/common/src/main/scala/org/lfdecentralizedtrust/splice/store/db/SpliceDbLockCounters.scala +++ b/apps/common/src/main/scala/org/lfdecentralizedtrust/splice/store/db/SpliceDbLockCounters.scala @@ -23,4 +23,6 @@ object SpliceDbLockCounters { val SCAN_WRITERS: DbLockCounter = DbLockCounter(105) val SPLITWELL_WRITE: DbLockCounter = DbLockCounter(106) val SPLITWELL_WRITERS: DbLockCounter = DbLockCounter(107) + val SYNC_OPERATOR_WRITE: DbLockCounter = DbLockCounter(108) + val SYNC_OPERATOR_WRITERS: DbLockCounter = DbLockCounter(109) } diff --git a/apps/common/src/test/scala/org/lfdecentralizedtrust/splice/store/StoreTestBase.scala b/apps/common/src/test/scala/org/lfdecentralizedtrust/splice/store/StoreTestBase.scala index 0290d7cb82..9411092f03 100644 --- a/apps/common/src/test/scala/org/lfdecentralizedtrust/splice/store/StoreTestBase.scala +++ b/apps/common/src/test/scala/org/lfdecentralizedtrust/splice/store/StoreTestBase.scala @@ -899,10 +899,11 @@ abstract class StoreTestBase domain: SynchronizerId, contract: Contract[TCid, T], counter: Long, + observers: Seq[PartyId] = Seq.empty, ): ActiveContract = ActiveContract( domain, - toCreatedEvent(contract, Seq(dsoParty)), + toCreatedEvent(contract, Seq(dsoParty), observers), counter, ) diff --git a/apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/store/db/DbSvDsoStore.scala b/apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/store/db/DbSvDsoStore.scala index 06d7133d6a..96518a71b9 100644 --- a/apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/store/db/DbSvDsoStore.scala +++ b/apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/store/db/DbSvDsoStore.scala @@ -42,6 +42,7 @@ import org.lfdecentralizedtrust.splice.store.db.{ AcsQueries, AcsTables, DbAppStore, + MemberTrafficQueries, StoreDescriptor, } import org.lfdecentralizedtrust.splice.store.{ @@ -121,6 +122,7 @@ class DbSvDsoStore( with SvDsoStore with AcsTables with AcsQueries + with MemberTrafficQueries with AcsJdbcTypes with DbVotesAcsStoreQueryBuilder with LimitHelpers { @@ -1649,25 +1651,14 @@ class DbSvDsoStore( override def getTotalPurchasedMemberTraffic(memberId: Member, synchronizerId: SynchronizerId)( implicit tc: TraceContext ): Future[Long] = waitUntilAcsIngested { - for { - sum <- storage - .querySingle( - sql""" - select sum(total_traffic_purchased) - from #${DsoTables.acsTableName} - where store_id = $acsStoreId - and migration_id = $domainMigrationId - and package_name = ${MemberTraffic.PACKAGE_NAME} - and template_id_qualified_name = ${QualifiedName( - MemberTraffic.TEMPLATE_ID_WITH_PACKAGE_ID - )} - and member_traffic_member = ${lengthLimited(memberId.toProtoPrimitive)} - and member_traffic_domain = $synchronizerId - """.as[Long].headOption, - "getTotalPurchasedMemberTraffic", - ) - .value - } yield sum.getOrElse(0L) + sumPurchasedMemberTraffic( + storage, + DsoTables.acsTableName, + acsStoreId, + domainMigrationId, + memberId, + synchronizerId, + ) } override def lookupVoteRequest( diff --git a/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/SyncOperatorApp.scala b/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/SyncOperatorApp.scala new file mode 100644 index 0000000000..30fcd6a1da --- /dev/null +++ b/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/SyncOperatorApp.scala @@ -0,0 +1,217 @@ +// Copyright (c) 2024 Digital Asset (Switzerland) GmbH and/or its affiliates. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +package org.lfdecentralizedtrust.splice.syncoperator + +import com.daml.grpc.adapter.ExecutionSequencerFactory +import com.digitalasset.canton.concurrent.FutureSupervisor +import com.digitalasset.canton.config.CantonRequireTypes.InstanceName +import com.digitalasset.canton.config.ProcessingTimeout +import com.digitalasset.canton.lifecycle.LifeCycle +import com.digitalasset.canton.logging.{NamedLoggerFactory, TracedLogger} +import com.digitalasset.canton.resource.DbStorage +import com.digitalasset.canton.time.Clock +import com.digitalasset.canton.topology.{PartyId, SynchronizerId} +import com.digitalasset.canton.tracing.{TraceContext, TracerProvider} +import io.grpc.Status +import io.opentelemetry.api.trace.Tracer +import org.apache.pekko.actor.ActorSystem +import org.lfdecentralizedtrust.splice.config.SharedSpliceAppParameters +import org.lfdecentralizedtrust.splice.environment.{ + BaseLedgerConnection, + Node, + PackageVersionSupport, + ParticipantAdminConnection, + RetryFor, + SequencerAdminConnection, + SpliceLedgerClient, +} +import org.lfdecentralizedtrust.splice.scan.admin.api.client.ScanConnection +import org.lfdecentralizedtrust.splice.syncoperator.automation.SyncOperatorAutomationService +import org.lfdecentralizedtrust.splice.syncoperator.config.SyncOperatorAppBackendConfig +import org.lfdecentralizedtrust.splice.syncoperator.metrics.SyncOperatorAppMetrics +import org.lfdecentralizedtrust.splice.syncoperator.store.SyncOperatorStore +import org.lfdecentralizedtrust.splice.util.HasHealth + +import scala.concurrent.{ExecutionContextExecutor, Future} + +/** Class representing a sync operator app instance. + * + * Ingests the traffic purchases made for the synchronizer it operates and grants them on that + * synchronizer's sequencer. + * + * Modelled after Canton's ParticipantNode class. + */ +class SyncOperatorApp( + override val name: InstanceName, + val config: SyncOperatorAppBackendConfig, + val appParameters: SharedSpliceAppParameters, + storage: DbStorage, + override protected val clock: Clock, + val loggerFactory: NamedLoggerFactory, + tracerProvider: TracerProvider, + futureSupervisor: FutureSupervisor, + metrics: SyncOperatorAppMetrics, +)(implicit + ac: ActorSystem, + ec: ExecutionContextExecutor, + esf: ExecutionSequencerFactory, + tracer: Tracer, +) extends Node[SyncOperatorApp.State, Unit]( + config.operatorUser, + config.participantClient, + appParameters, + loggerFactory, + tracerProvider, + futureSupervisor, + metrics, + ) { + + override lazy val ports = Map("admin" -> config.adminApi.port) + + override def preInitializeAfterLedgerConnection( + connection: BaseLedgerConnection, + ledgerClient: SpliceLedgerClient, + )(implicit traceContext: TraceContext): Future[Unit] = Future.unit + + override def initialize( + ledgerClient: SpliceLedgerClient, + partyId: PartyId, + preInitializeState: Unit, + )(implicit traceContext: TraceContext): Future[SyncOperatorApp.State] = { + for { + scanConnection <- appInitStep(s"Get scan connection") { + ScanConnection.singleCached( + ledgerClient, + config.scanClient, + appParameters.upgradesConfig, + clock, + retryProvider, + loggerFactory, + ) + } + participantAdminConnection = new ParticipantAdminConnection( + config.participantClient.adminApi, + appParameters.loggingConfig.api, + loggerFactory, + metrics.grpcClientMetrics, + retryProvider, + ) + participantId <- appInitStep("Get participant id") { + participantAdminConnection.getParticipantId() + } + dsoParty <- appInitStep("Get DSO party id") { scanConnection.getDsoPartyId() } + sequencerAdminConnection = new SequencerAdminConnection( + config.sequencer.adminApi, + appParameters.loggingConfig.api, + loggerFactory, + metrics.grpcClientMetrics, + retryProvider, + ) + synchronizerId <- appInitStep("Get the synchronizer id from the sequencer") { + servedSynchronizerId(sequencerAdminConnection) + } + storeKey = SyncOperatorStore.Key( + operatorParty = partyId, + dsoParty = dsoParty, + synchronizerId = synchronizerId, + ) + store = SyncOperatorStore( + storeKey, + storage, + loggerFactory, + retryProvider, + // MIGRATION_ID is frozen network-wide and logical synchronizer upgrades carry a serial id + // instead, so the store's partition never has to move. + 0L, + participantId, + config.automation.ingestion, + config.parameters.defaultLimit, + ) + globalSynchronizerId <- appInitStep("Get the global synchronizer id") { + scanConnection.getAmuletRulesDomain()(traceContext) + } + readOnlyLedgerConnection = ledgerClient + .readOnlyConnection( + this.getClass.getSimpleName, + loggerFactory, + ) + packageVersionSupport = PackageVersionSupport.createPackageVersionSupport( + globalSynchronizerId, + readOnlyLedgerConnection, + loggerFactory, + ) + automation = new SyncOperatorAutomationService( + config.automation, + clock, + store, + storage, + ledgerClient, + retryProvider, + config.parameters, + loggerFactory, + packageVersionSupport, + ) + } yield { + SyncOperatorApp.State( + automation, + storage, + store, + scanConnection, + participantAdminConnection, + sequencerAdminConnection, + loggerFactory.getTracedLogger(SyncOperatorApp.State.getClass), + timeouts, + ) + } + } + + /** The synchronizer the configured sequencer serves. Waits while it is still initializing. */ + private def servedSynchronizerId( + sequencerAdminConnection: SequencerAdminConnection + )(implicit traceContext: TraceContext): Future[SynchronizerId] = + retryProvider.getValueWithRetries( + RetryFor.WaitingOnInitDependency, + "sync_operator_served_synchronizer_id", + "the sequencer reports the synchronizer it serves", + sequencerAdminConnection.getStatus.map( + _.successOption + .map(_.synchronizerId.logical) + .getOrElse( + throw Status.UNAVAILABLE + .withDescription("Sequencer is not yet initialized") + .asRuntimeException() + ) + ), + logger, + ) + + protected[this] override def automationServices(st: SyncOperatorApp.State) = + Seq(st.automation) +} + +object SyncOperatorApp { + case class State( + automation: SyncOperatorAutomationService, + storage: DbStorage, + store: SyncOperatorStore, + scanConnection: ScanConnection, + participantAdminConnection: ParticipantAdminConnection, + sequencerAdminConnection: SequencerAdminConnection, + logger: TracedLogger, + timeouts: ProcessingTimeout, + ) extends AutoCloseable + with HasHealth { + override def isHealthy: Boolean = storage.isActive + + override def close(): Unit = + LifeCycle.close( + automation, + storage, + store, + scanConnection, + participantAdminConnection, + sequencerAdminConnection, + )(logger) + } +} diff --git a/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/SyncOperatorAppBootstrap.scala b/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/SyncOperatorAppBootstrap.scala new file mode 100644 index 0000000000..53d5ba2ee9 --- /dev/null +++ b/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/SyncOperatorAppBootstrap.scala @@ -0,0 +1,137 @@ +// Copyright (c) 2024 Digital Asset (Switzerland) GmbH and/or its affiliates. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +package org.lfdecentralizedtrust.splice.syncoperator + +import cats.data.EitherT +import cats.syntax.either.* +import com.daml.grpc.adapter.ExecutionSequencerFactory +import com.digitalasset.canton.concurrent.{ + ExecutionContextIdlenessExecutorService, + FutureSupervisor, +} +import com.digitalasset.canton.config.CantonRequireTypes.InstanceName +import com.digitalasset.canton.config.TestingConfigInternal +import com.digitalasset.canton.logging.NamedLoggerFactory +import com.digitalasset.canton.resource.* +import com.digitalasset.canton.telemetry.ConfiguredOpenTelemetry +import com.digitalasset.canton.time.* +import org.apache.pekko.actor.ActorSystem +import org.lfdecentralizedtrust.splice.admin.http.AdminRoutes +import org.lfdecentralizedtrust.splice.config.SharedSpliceAppParameters +import org.lfdecentralizedtrust.splice.config.SpliceDbConfig.withConfiguredPostgresConnectionSettings +import org.lfdecentralizedtrust.splice.environment.{NodeBootstrapBase, SpliceStorageFactory} +import org.lfdecentralizedtrust.splice.store.db.SpliceDbLockCounters +import org.lfdecentralizedtrust.splice.syncoperator.config.SyncOperatorAppBackendConfig +import org.lfdecentralizedtrust.splice.syncoperator.metrics.SyncOperatorAppMetrics + +import java.util.concurrent.ScheduledExecutorService + +import scala.concurrent.Future + +/** Class used to orchester the starting/initialization of Sync Operator Node apps. + * + * Modelled after Canton's ParticipantNodeBootstrap class. + */ +class SyncOperatorAppBootstrap( + override val name: InstanceName, + val config: SyncOperatorAppBackendConfig, + val syncOperatorAppParameters: SharedSpliceAppParameters, + val testingConfig: TestingConfigInternal, + clock: Clock, + override val metrics: SyncOperatorAppMetrics, + storageFactory: StorageFactory, + loggerFactory: NamedLoggerFactory, + futureSupervisor: FutureSupervisor, + configuredOpenTelemetry: ConfiguredOpenTelemetry, +)(implicit + executionContext: ExecutionContextIdlenessExecutorService, + scheduler: ScheduledExecutorService, + actorSystem: ActorSystem, + executionSequencerFactory: ExecutionSequencerFactory, +) extends NodeBootstrapBase[ + SyncOperatorApp, + SyncOperatorAppBackendConfig, + SharedSpliceAppParameters, + ]( + config, + name, + syncOperatorAppParameters, + clock, + metrics, + storageFactory, + loggerFactory, + configuredOpenTelemetry, + ) { + + override def initialize(adminRoutes: AdminRoutes): EitherT[Future, String, Unit] = { + // No HTTP surface yet; operator-facing endpoints attach here. + val _ = adminRoutes + startInstanceUnlessClosing { + new SyncOperatorApp( + name, + config, + syncOperatorAppParameters, + storage, + clock, + loggerFactory, + tracerProvider, + futureSupervisor, + metrics, + ) + } + } + + override def isActive: Boolean = storage.isActive +} + +object SyncOperatorAppBootstrap { + val LoggerFactoryKeyName: String = "syncoperator" + + def apply( + name: String, + syncOperatorConfig: SyncOperatorAppBackendConfig, + syncOperatorAppParameters: SharedSpliceAppParameters, + clock: Clock, + syncOperatorMetrics: SyncOperatorAppMetrics, + testingConfigInternal: TestingConfigInternal, + futureSupervisor: FutureSupervisor, + loggerFactory: NamedLoggerFactory, + configuredOpenTelemetry: ConfiguredOpenTelemetry, + )(implicit + executionContext: ExecutionContextIdlenessExecutorService, + scheduler: ScheduledExecutorService, + actorSystem: ActorSystem, + executionSequencerFactory: ExecutionSequencerFactory, + ): Either[String, SyncOperatorAppBootstrap] = + SpliceStorageFactory.createWithDeferredClose( + storage = withConfiguredPostgresConnectionSettings( + syncOperatorConfig.storage, + syncOperatorConfig.postgres, + ), + instanceLockEnabled = syncOperatorConfig.instanceLockEnabled, + mainLockCounter = SpliceDbLockCounters.SYNC_OPERATOR_WRITE, + poolLockCounter = SpliceDbLockCounters.SYNC_OPERATOR_WRITERS, + exitOnFatalFailures = syncOperatorAppParameters.exitOnFatalFailures, + futureSupervisor = futureSupervisor, + loggerFactory = loggerFactory, + ) { storageFactory => + InstanceName + .create(name) + .map { instanceName => + new SyncOperatorAppBootstrap( + instanceName, + syncOperatorConfig, + syncOperatorAppParameters, + testingConfigInternal, + clock, + syncOperatorMetrics, + storageFactory, + loggerFactory, + futureSupervisor, + configuredOpenTelemetry, + ) + } + .leftMap(_.toString) + } +} diff --git a/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/automation/SyncOperatorAutomationService.scala b/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/automation/SyncOperatorAutomationService.scala new file mode 100644 index 0000000000..3821e1e148 --- /dev/null +++ b/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/automation/SyncOperatorAutomationService.scala @@ -0,0 +1,68 @@ +// Copyright (c) 2024 Digital Asset (Switzerland) GmbH and/or its affiliates. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +package org.lfdecentralizedtrust.splice.syncoperator.automation + +import com.digitalasset.canton.logging.NamedLoggerFactory +import com.digitalasset.canton.resource.DbStorage +import com.digitalasset.canton.time.Clock +import io.opentelemetry.api.trace.Tracer +import org.apache.pekko.stream.Materializer +import org.lfdecentralizedtrust.splice.automation.{ + AutomationServiceCompanion, + SpliceAppAutomationService, + SqlIndexInitializationTrigger, +} +import org.lfdecentralizedtrust.splice.config.{AutomationConfig, SpliceParametersConfig} +import org.lfdecentralizedtrust.splice.environment.{ + PackageVersionSupport, + RetryProvider, + SpliceLedgerClient, +} +import org.lfdecentralizedtrust.splice.store.DomainTimeSynchronization +import org.lfdecentralizedtrust.splice.syncoperator.store.SyncOperatorStore + +import scala.concurrent.ExecutionContextExecutor + +/** Manages background automation that runs on a sync operator app. */ +class SyncOperatorAutomationService( + automationConfig: AutomationConfig, + clock: Clock, + override val store: SyncOperatorStore, + storage: DbStorage, + ledgerClient: SpliceLedgerClient, + retryProvider: RetryProvider, + params: SpliceParametersConfig, + protected val loggerFactory: NamedLoggerFactory, + packageVersionSupport: PackageVersionSupport, +)(implicit + ec: ExecutionContextExecutor, + mat: Materializer, + tracer: Tracer, +) extends SpliceAppAutomationService( + automationConfig, + clock, + // Nothing registered here depends on domain time. + DomainTimeSynchronization.Noop, + store, + ledgerClient, + retryProvider, + params, + packageVersionSupport, + ) { + + override def companion: SyncOperatorAutomationService.type = SyncOperatorAutomationService + + registerTrigger( + SqlIndexInitializationTrigger( + storage, + triggerContext, + ) + ) +} + +object SyncOperatorAutomationService extends AutomationServiceCompanion { + + override protected[this] def expectedTriggerClasses: Seq[Nothing] = + Seq.empty +} diff --git a/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/config/SyncOperatorAppConfig.scala b/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/config/SyncOperatorAppConfig.scala new file mode 100644 index 0000000000..0204f28a27 --- /dev/null +++ b/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/config/SyncOperatorAppConfig.scala @@ -0,0 +1,48 @@ +// Copyright (c) 2024 Digital Asset (Switzerland) GmbH and/or its affiliates. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +package org.lfdecentralizedtrust.splice.syncoperator.config + +import com.digitalasset.canton.config.* +import org.lfdecentralizedtrust.splice.config.{ + AutomationConfig, + HttpClientConfig, + NetworkAppClientConfig, + ParticipantClientConfig, + SpliceBackendConfig, + SpliceParametersConfig, + SplicePostgresConfig, +} +import org.lfdecentralizedtrust.splice.scan.config.ScanAppClientConfig + +// The sequencer this node grants traffic on. +case class SyncOperatorSequencerConfig( + adminApi: FullClientConfig +) + +case class SyncOperatorAppBackendConfig( + override val adminApi: AdminServerConfig = AdminServerConfig(), + override val storage: DbConfig, + postgres: SplicePostgresConfig = SplicePostgresConfig(), + // Ledger API user of the operator party. + operatorUser: String, + participantClient: ParticipantClientConfig, + scanClient: ScanAppClientConfig, + sequencer: SyncOperatorSequencerConfig, + override val automation: AutomationConfig = AutomationConfig(), + parameters: SpliceParametersConfig = SpliceParametersConfig(batching = BatchingConfig()), + // Set to false to disable the DB-level exclusive lock that prevents two sync operator instances + // from running concurrently against the same database. Only disable for migration scenarios + // where intentional overlap is required. + instanceLockEnabled: Boolean = true, +) extends SpliceBackendConfig { + override val nodeTypeName: String = "syncoperator" + + override def clientAdminApi: ClientConfig = adminApi.clientConfig +} + +case class SyncOperatorAppClientConfig( + adminApi: NetworkAppClientConfig +) extends HttpClientConfig { + override def clientAdminApi: NetworkAppClientConfig = adminApi +} diff --git a/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/metrics/SyncOperatorAppMetrics.scala b/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/metrics/SyncOperatorAppMetrics.scala new file mode 100644 index 0000000000..97607f8f5d --- /dev/null +++ b/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/metrics/SyncOperatorAppMetrics.scala @@ -0,0 +1,15 @@ +// Copyright (c) 2024 Digital Asset (Switzerland) GmbH and/or its affiliates. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +package org.lfdecentralizedtrust.splice.syncoperator.metrics + +import com.daml.metrics.api.MetricHandle.LabeledMetricsFactory +import com.digitalasset.canton.logging.NamedLoggerFactory +import com.digitalasset.canton.metrics.DbStorageHistograms +import org.lfdecentralizedtrust.splice.BaseSpliceMetrics + +class SyncOperatorAppMetrics( + metricsFactory: LabeledMetricsFactory, + storageHistograms: DbStorageHistograms, + loggerFactory: NamedLoggerFactory, +) extends BaseSpliceMetrics("syncoperator", metricsFactory, storageHistograms, loggerFactory) {} diff --git a/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/store/SyncOperatorStore.scala b/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/store/SyncOperatorStore.scala new file mode 100644 index 0000000000..6e0756994b --- /dev/null +++ b/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/store/SyncOperatorStore.scala @@ -0,0 +1,117 @@ +// Copyright (c) 2024 Digital Asset (Switzerland) GmbH and/or its affiliates. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +package org.lfdecentralizedtrust.splice.syncoperator.store + +import com.digitalasset.canton.lifecycle.CloseContext +import com.digitalasset.canton.logging.NamedLoggerFactory +import com.digitalasset.canton.logging.pretty.{Pretty, PrettyPrinting} +import com.digitalasset.canton.resource.DbStorage +import com.digitalasset.canton.topology.{Member, ParticipantId, PartyId, SynchronizerId} +import com.digitalasset.canton.tracing.TraceContext +import org.lfdecentralizedtrust.splice.codegen.java.splice +import org.lfdecentralizedtrust.splice.config.IngestionConfig +import org.lfdecentralizedtrust.splice.environment.RetryProvider +import org.lfdecentralizedtrust.splice.store.db.AcsInterfaceViewRowData +import org.lfdecentralizedtrust.splice.store.{AppStore, Limit, MultiDomainAcsStore} +import org.lfdecentralizedtrust.splice.syncoperator.store.db.DbSyncOperatorStore +import org.lfdecentralizedtrust.splice.syncoperator.store.db.SyncOperatorTables.SyncOperatorAcsStoreRowData +import org.lfdecentralizedtrust.splice.util.TemplateJsonDecoder + +import scala.concurrent.{ExecutionContext, Future} +import scala.jdk.OptionConverters.* + +/** Store of a sync operator app. + * + * Ingests the `MemberTraffic` purchases made for this operator's synchronizer, which the buy choice + * makes it an observer of. + */ +trait SyncOperatorStore extends AppStore { + + /** The parties and synchronizer this store is scoped to. */ + val key: SyncOperatorStore.Key + + override def multiDomainAcsStore: MultiDomainAcsStore + + /** Total traffic purchased for `memberId` on this operator's synchronizer. */ + def getTotalPurchasedMemberTraffic(memberId: Member)(implicit + tc: TraceContext + ): Future[Long] +} + +object SyncOperatorStore { + + def apply( + key: Key, + storage: DbStorage, + loggerFactory: NamedLoggerFactory, + retryProvider: RetryProvider, + migrationId: Long, + participantId: ParticipantId, + ingestionConfig: IngestionConfig, + defaultLimit: Limit, + )(implicit + ec: ExecutionContext, + templateJsonDecoder: TemplateJsonDecoder, + close: CloseContext, + ): SyncOperatorStore = + new DbSyncOperatorStore( + key, + storage, + loggerFactory, + retryProvider, + migrationId, + participantId, + ingestionConfig, + defaultLimit, + ) + + case class Key( + /** The registered operator of [[synchronizerId]]. */ + operatorParty: PartyId, + /** The DSO party, sole signatory of `MemberTraffic`. */ + dsoParty: PartyId, + /** The dedicated synchronizer this operator serves. */ + synchronizerId: SynchronizerId, + ) extends PrettyPrinting { + override def pretty: Pretty[Key] = prettyOfClass( + param("operatorParty", _.operatorParty), + param("dsoParty", _.dsoParty), + param("synchronizerId", _.synchronizerId), + ) + } + + def contractFilter(key: Key): MultiDomainAcsStore.ContractFilter[ + SyncOperatorAcsStoreRowData, + AcsInterfaceViewRowData.NoInterfacesIngested, + ] = { + import MultiDomainAcsStore.mkFilter + val operator = key.operatorParty.toProtoPrimitive + val dso = key.dsoParty.toProtoPrimitive + val synchronizerId = key.synchronizerId.toProtoPrimitive + + MultiDomainAcsStore.SimpleContractFilter( + key.operatorParty, + Map( + mkFilter(splice.decentralizedsynchronizer.MemberTraffic.COMPANION)(co => + co.payload.dso == dso && + co.payload.operator.toScala.contains(operator) && + co.payload.synchronizerId == synchronizerId && + co.payload.migrationId == 0L // A registered synchronizer is pinned to migration id 0. + ) { contract => + SyncOperatorAcsStoreRowData( + contract, + memberTrafficMember = Member + .fromProtoPrimitive_(contract.payload.memberId) + // we ignore cases where the member id is invalid instead of throwing an exception + // to avoid killing the entire ingestion pipeline as a result + .fold(_ => None, Some(_)), + // the filter has already established these agree, so avoid parsing per contract + memberTrafficDomain = Some(key.synchronizerId), + totalTrafficPurchased = Some(contract.payload.totalPurchased), + ) + } + ), + ) + } +} diff --git a/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/store/db/DbSyncOperatorStore.scala b/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/store/db/DbSyncOperatorStore.scala new file mode 100644 index 0000000000..8c8f126cb7 --- /dev/null +++ b/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/store/db/DbSyncOperatorStore.scala @@ -0,0 +1,97 @@ +// Copyright (c) 2024 Digital Asset (Switzerland) GmbH and/or its affiliates. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +package org.lfdecentralizedtrust.splice.syncoperator.store.db + +import com.digitalasset.canton.lifecycle.CloseContext +import com.digitalasset.canton.logging.NamedLoggerFactory +import com.digitalasset.canton.resource.DbStorage +import com.digitalasset.canton.topology.{Member, ParticipantId, PartyId} +import com.digitalasset.canton.tracing.TraceContext +import org.lfdecentralizedtrust.splice.config.IngestionConfig +import org.lfdecentralizedtrust.splice.environment.RetryProvider +import org.lfdecentralizedtrust.splice.store.db.AcsQueries.AcsStoreId +import org.lfdecentralizedtrust.splice.store.db.{ + AcsInterfaceViewRowData, + AcsQueries, + AcsTables, + DbAppStore, + MemberTrafficQueries, + StoreDescriptor, +} +import org.lfdecentralizedtrust.splice.store.{Limit, LimitHelpers, MultiDomainAcsStore} +import org.lfdecentralizedtrust.splice.syncoperator.store.SyncOperatorStore +import org.lfdecentralizedtrust.splice.syncoperator.store.db.SyncOperatorTables.SyncOperatorAcsStoreRowData +import org.lfdecentralizedtrust.splice.util.TemplateJsonDecoder + +import scala.concurrent.{ExecutionContext, Future} + +class DbSyncOperatorStore( + override val key: SyncOperatorStore.Key, + storage: DbStorage, + override protected val loggerFactory: NamedLoggerFactory, + override protected val retryProvider: RetryProvider, + domainMigrationId: Long, + participantId: ParticipantId, + ingestionConfig: IngestionConfig, + override val defaultLimit: Limit, +)(implicit + override protected val ec: ExecutionContext, + templateJsonDecoder: TemplateJsonDecoder, + closeContext: CloseContext, +) extends DbAppStore( + storage = storage, + acsTableName = SyncOperatorTables.acsTableName, + interfaceViewsTableNameOpt = None, + // Any change in the store descriptor will lead to previously deployed applications + // forgetting all persisted data once they upgrade to the new version. + // WARNING: Reinitializing the acs store is a very expensive operation, as it currently fetches the full + // unfiltered ACS from the participant, irrespective of the filter defined by `acsContractFilter`. + // This may lead to the entire app being unavailable or not working properly until the full ACS has been ingested. + // Do not modify any part of the store descriptor unless you are sure that the resulting downtime is acceptable. + // If you do modify it, make sure to very clearly document in the release notes that there will be planned downtime, + // and notify the person coordinating the deployment. + acsStoreDescriptor = StoreDescriptor( + version = 1, + name = "DbSyncOperatorStore", + party = key.operatorParty, + participant = participantId, + key = Map( + "operatorParty" -> key.operatorParty.toProtoPrimitive, + "dsoParty" -> key.dsoParty.toProtoPrimitive, + "synchronizerId" -> key.synchronizerId.toProtoPrimitive, + ), + ), + migrationId = domainMigrationId, + ingestionConfig, + ) + with AcsTables + with AcsQueries + with LimitHelpers + with MemberTrafficQueries + with SyncOperatorStore { + + override def dsoPartyId: PartyId = key.dsoParty + + override lazy val acsContractFilter: MultiDomainAcsStore.ContractFilter[ + SyncOperatorAcsStoreRowData, + AcsInterfaceViewRowData.NoInterfacesIngested, + ] = SyncOperatorStore.contractFilter(key) + + import multiDomainAcsStore.waitUntilAcsIngested + + private def acsStoreId: AcsStoreId = multiDomainAcsStore.acsStoreId + + override def getTotalPurchasedMemberTraffic(memberId: Member)(implicit + tc: TraceContext + ): Future[Long] = waitUntilAcsIngested { + sumPurchasedMemberTraffic( + storage, + SyncOperatorTables.acsTableName, + acsStoreId, + domainMigrationId, + memberId, + key.synchronizerId, + ) + } +} diff --git a/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/store/db/SyncOperatorTables.scala b/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/store/db/SyncOperatorTables.scala new file mode 100644 index 0000000000..8c517526ba --- /dev/null +++ b/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/store/db/SyncOperatorTables.scala @@ -0,0 +1,48 @@ +// Copyright (c) 2024 Digital Asset (Switzerland) GmbH and/or its affiliates. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +package org.lfdecentralizedtrust.splice.syncoperator.store.db + +import com.digitalasset.canton.topology.{Member, SynchronizerId} +import com.digitalasset.daml.lf.data.Time.Timestamp +import org.lfdecentralizedtrust.splice.store.db.AcsRowData.HasIndexColumns +import org.lfdecentralizedtrust.splice.store.db.{AcsRowData, AcsTables, IndexColumnValue} +import org.lfdecentralizedtrust.splice.util.Contract + +object SyncOperatorTables extends AcsTables { + + case class SyncOperatorAcsStoreRowData( + contract: Contract[?, ?], + contractExpiresAt: Option[Timestamp] = None, + memberTrafficMember: Option[Member] = None, + memberTrafficDomain: Option[SynchronizerId] = None, + totalTrafficPurchased: Option[Long] = None, + ) extends AcsRowData.AcsRowDataFromContract { + override def indexColumns: Seq[(String, IndexColumnValue[?])] = + Seq( + SyncOperatorAcsStoreRowData.IndexColumns.member_traffic_member -> memberTrafficMember, + SyncOperatorAcsStoreRowData.IndexColumns.member_traffic_domain -> memberTrafficDomain, + SyncOperatorAcsStoreRowData.IndexColumns.total_traffic_purchased -> totalTrafficPurchased, + ) + } + + object SyncOperatorAcsStoreRowData { + implicit val hasIndexColumns: HasIndexColumns[SyncOperatorAcsStoreRowData] = + new HasIndexColumns[SyncOperatorAcsStoreRowData] { + override def indexColumnNames: Seq[String] = IndexColumns.All + } + private object IndexColumns { + val member_traffic_member = "member_traffic_member" + val member_traffic_domain = "member_traffic_domain" + val total_traffic_purchased = "total_traffic_purchased" + + val All: Seq[String] = Seq( + member_traffic_member, + member_traffic_domain, + total_traffic_purchased, + ) + } + } + + val acsTableName = "sync_operator_acs_store" +} diff --git a/apps/syncoperator/src/test/scala/org/lfdecentralizedtrust/splice/store/db/SyncOperatorStoreTest.scala b/apps/syncoperator/src/test/scala/org/lfdecentralizedtrust/splice/store/db/SyncOperatorStoreTest.scala new file mode 100644 index 0000000000..3e2e038659 --- /dev/null +++ b/apps/syncoperator/src/test/scala/org/lfdecentralizedtrust/splice/store/db/SyncOperatorStoreTest.scala @@ -0,0 +1,181 @@ +// Copyright (c) 2024 Digital Asset (Switzerland) GmbH and/or its affiliates. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +package org.lfdecentralizedtrust.splice.store.db + +import com.daml.metrics.api.noop.NoOpMetricsFactory +import com.digitalasset.canton.{HasActorSystem, HasExecutionContext, SynchronizerAlias} +import com.digitalasset.canton.concurrent.FutureSupervisor +import com.digitalasset.canton.lifecycle.FutureUnlessShutdown +import com.digitalasset.canton.resource.DbStorage +import com.digitalasset.canton.tracing.TraceContext +import com.digitalasset.canton.topology.{Member, PartyId, SynchronizerId} +import org.lfdecentralizedtrust.splice.codegen.java.splice.decentralizedsynchronizer.MemberTraffic +import org.lfdecentralizedtrust.splice.config.IngestionConfig +import org.lfdecentralizedtrust.splice.environment.{DarResources, RetryProvider} +import org.lfdecentralizedtrust.splice.store.{HardLimit, Limit, StoreTestBase} +import org.lfdecentralizedtrust.splice.syncoperator.store.SyncOperatorStore +import org.lfdecentralizedtrust.splice.syncoperator.store.db.DbSyncOperatorStore +import org.lfdecentralizedtrust.splice.util.{Contract, ResourceTemplateDecoder, TemplateJsonDecoder} + +import scala.concurrent.Future +import scala.jdk.OptionConverters.* + +abstract class SyncOperatorStoreTest extends StoreTestBase with HasExecutionContext { + + protected val operatorParty: PartyId = providerParty(1) + protected val otherOperator: PartyId = providerParty(2) + protected val ourSynchronizer: SynchronizerId = + SynchronizerId.tryFromString("dedicated::operator") + protected val foreignSynchronizer: SynchronizerId = + SynchronizerId.tryFromString("dedicated::someone-else") + + protected val alice: Member = mkParticipantId("alice") + protected val bob: Member = mkParticipantId("bob") + + protected def mkStore(): Future[SyncOperatorStore] + + private def memberTraffic( + member: Member, + totalPurchased: Long, + synchronizerId: SynchronizerId = ourSynchronizer, + operator: Option[PartyId] = Some(operatorParty), + migrationId: Long = 0L, + ): Contract[MemberTraffic.ContractId, MemberTraffic] = { + val template = new MemberTraffic( + dsoParty.toProtoPrimitive, + member.toProtoPrimitive, + synchronizerId.toProtoPrimitive, + migrationId, + totalPurchased, + 1L, + java.math.BigDecimal.ONE, + java.math.BigDecimal.ONE, + operator.map(_.toProtoPrimitive).toJava, + ) + contract( + MemberTraffic.TEMPLATE_ID_WITH_PACKAGE_ID, + new MemberTraffic.ContractId(nextCid()), + template, + ) + } + + private def ingest( + store: SyncOperatorStore, + contracts: Seq[Contract[MemberTraffic.ContractId, MemberTraffic]], + ): Future[Unit] = + store.multiDomainAcsStore.testIngestionSink.ingestAcs( + nextOffset(), + // the purchase is made and recorded on the decentralized synchronizer, not on the dedicated + // one it names + contracts.zipWithIndex.map { case (c, i) => + // mirrors `observer (optionalToList operator)` on the template + val observers = c.payload.operator.toScala.map(PartyId.tryFromProtoPrimitive).toList + toActiveContract(dummyDomain, c, i.toLong, observers) + }, + Seq.empty, + Seq.empty, + ) + + "SyncOperatorStore" should { + + "sum the purchases made for this synchronizer" in { + for { + store <- mkStore() + _ <- ingest(store, Seq(memberTraffic(alice, 100L), memberTraffic(alice, 250L))) + total <- store.getTotalPurchasedMemberTraffic(alice) + } yield total shouldBe 350L + } + + "keep members apart" in { + for { + store <- mkStore() + _ <- ingest(store, Seq(memberTraffic(alice, 100L), memberTraffic(bob, 700L))) + aliceTotal <- store.getTotalPurchasedMemberTraffic(alice) + bobTotal <- store.getTotalPurchasedMemberTraffic(bob) + } yield { + aliceTotal shouldBe 100L + bobTotal shouldBe 700L + } + } + + "ignore purchases for another synchronizer" in { + for { + store <- mkStore() + _ <- ingest(store, Seq(memberTraffic(alice, 100L, synchronizerId = foreignSynchronizer))) + total <- store.getTotalPurchasedMemberTraffic(alice) + } yield total shouldBe 0L + } + + "ignore purchases observed by another operator" in { + for { + store <- mkStore() + _ <- ingest(store, Seq(memberTraffic(alice, 100L, operator = Some(otherOperator)))) + total <- store.getTotalPurchasedMemberTraffic(alice) + } yield total shouldBe 0L + } + + "ignore purchases with no operator, which are decentralized-synchronizer traffic" in { + for { + store <- mkStore() + _ <- ingest(store, Seq(memberTraffic(alice, 100L, operator = None))) + total <- store.getTotalPurchasedMemberTraffic(alice) + } yield total shouldBe 0L + } + + // A registered synchronizer is pinned to migration id 0, so anything else is not ours even if + // it names our synchronizer and operator. + "ignore purchases carrying a non-zero migration id" in { + for { + store <- mkStore() + _ <- ingest(store, Seq(memberTraffic(alice, 100L, migrationId = 1L))) + total <- store.getTotalPurchasedMemberTraffic(alice) + } yield total shouldBe 0L + } + + "report zero for a member that has never purchased" in { + for { + store <- mkStore() + // queries wait on ACS ingestion, so the empty ACS still has to be ingested + _ <- ingest(store, Seq.empty) + total <- store.getTotalPurchasedMemberTraffic(alice) + } yield total shouldBe 0L + } + } +} + +class DbSyncOperatorStoreTest + extends SyncOperatorStoreTest + with HasActorSystem + with SplicePostgresTest + with AcsJdbcTypes + with AcsTables { + + override protected def mkStore(): Future[SyncOperatorStore] = { + val packageSignatures = + ResourceTemplateDecoder.loadPackageSignaturesFromResources(DarResources.amulet.all) + implicit val templateJsonDecoder: TemplateJsonDecoder = + new ResourceTemplateDecoder(packageSignatures, loggerFactory) + + val store = new DbSyncOperatorStore( + SyncOperatorStore.Key(operatorParty, dsoParty, ourSynchronizer), + storage, + loggerFactory, + RetryProvider(loggerFactory, timeouts, FutureSupervisor.Noop, NoOpMetricsFactory), + domainMigrationId, + participantId = mkParticipantId("SyncOperatorStoreTest"), + IngestionConfig(), + defaultLimit = HardLimit.tryCreate(Limit.DefaultMaxPageSize), + )(parallelExecutionContext, implicitly, implicitly) + for { + _ <- store.multiDomainAcsStore.testIngestionSink.initialize() + _ <- store.domains.ingestionSink.ingestConnectedDomains( + Map(SynchronizerAlias.tryCreate(dummyDomain.toProtoPrimitive) -> dummyDomain) + ) + } yield store + } + + override protected def cleanDb( + storage: DbStorage + )(implicit traceContext: TraceContext): FutureUnlessShutdown[?] = resetAllAppTables(storage) +} diff --git a/build.sbt b/build.sbt index bc7e72cc19..5804ad1e0e 100644 --- a/build.sbt +++ b/build.sbt @@ -85,6 +85,7 @@ lazy val root: Project = (project in file(".")) `apps-validator`, `apps-scan`, `apps-splitwell`, + `apps-syncoperator`, `apps-sv`, `apps-app`, `apps-metrics-docs`, @@ -1897,6 +1898,17 @@ lazy val `apps-splitwell` = ), ) +lazy val `apps-syncoperator` = + project + .in(file("apps/syncoperator")) + .dependsOn( + `apps-common` % "compile->compile;test->test", + `apps-scan` % "compile->compile;test->test", + ) + .settings( + BuildCommon.sharedAppSettings, + ) + lazy val pulumi = project .in(file("cluster/pulumi")) diff --git a/project/BuildCommon.scala b/project/BuildCommon.scala index 4aff0be830..7dacfc720f 100644 --- a/project/BuildCommon.scala +++ b/project/BuildCommon.scala @@ -246,6 +246,7 @@ object BuildCommon { "apps-validator/clean", "apps-scan/clean", "apps-splitwell/clean", + "apps-syncoperator/clean", "apps-sv/clean", "apps-wallet/clean", "apps-app/clean", diff --git a/test-full-class-names-non-integration.log b/test-full-class-names-non-integration.log index 6e5afedcd2..b3b38fee3e 100644 --- a/test-full-class-names-non-integration.log +++ b/test-full-class-names-non-integration.log @@ -42,6 +42,7 @@ org.lfdecentralizedtrust.splice.store.db.DbScanRewardsReferenceStoreTest org.lfdecentralizedtrust.splice.store.db.DbScanStoreTest org.lfdecentralizedtrust.splice.store.db.DbSvDsoStoreTest org.lfdecentralizedtrust.splice.store.db.DbSvSvStoreTest +org.lfdecentralizedtrust.splice.store.db.DbSyncOperatorStoreTest org.lfdecentralizedtrust.splice.store.db.DbTcsStoreTest org.lfdecentralizedtrust.splice.store.db.DbUserWalletStoreTest org.lfdecentralizedtrust.splice.store.db.SpliceStorageMultiLockTest