From cdf23be7d955411d7e4c4b99b208cd925b59a37b Mon Sep 17 00:00:00 2001 From: sadiq1971 Date: Thu, 20 Aug 2026 04:46:07 +0600 Subject: [PATCH 1/5] feat: sync operator app skeleton (Scala) [ci] Adds the apps/syncoperator module: config, store, automation host and health, ingesting the MemberTraffic purchases the operator observes for its own synchronizer and holding the sequencer admin connection the reconciliation will grant on. Ingestion is deliberately not filtered by the node's own migration id, since a registered synchronizer is pinned to migration id 0. Signed-off-by: sadiq1971 --- .../stable/V073__sync_operator_acs_store.sql | 24 ++ .../store/db/SpliceDbLockCounters.scala | 2 + .../splice/syncoperator/SyncOperatorApp.scala | 260 ++++++++++++++++++ .../SyncOperatorAppBootstrap.scala | 138 ++++++++++ .../SyncOperatorAutomationService.scala | 73 +++++ .../config/SyncOperatorAppConfig.scala | 60 ++++ .../metrics/SyncOperatorAppMetrics.scala | 15 + .../store/SyncOperatorStore.scala | 125 +++++++++ .../store/db/DbSyncOperatorStore.scala | 109 ++++++++ .../store/db/SyncOperatorTables.scala | 48 ++++ .../store/db/SyncOperatorStoreTest.scala | 177 ++++++++++++ build.sbt | 13 + 12 files changed, 1044 insertions(+) create mode 100644 apps/common/src/main/resources/db/migration/canton-network/postgres/stable/V073__sync_operator_acs_store.sql create mode 100644 apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/SyncOperatorApp.scala create mode 100644 apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/SyncOperatorAppBootstrap.scala create mode 100644 apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/automation/SyncOperatorAutomationService.scala create mode 100644 apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/config/SyncOperatorAppConfig.scala create mode 100644 apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/metrics/SyncOperatorAppMetrics.scala create mode 100644 apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/store/SyncOperatorStore.scala create mode 100644 apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/store/db/DbSyncOperatorStore.scala create mode 100644 apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/store/db/SyncOperatorTables.scala create mode 100644 apps/syncoperator/src/test/scala/org/lfdecentralizedtrust/splice/store/db/SyncOperatorStoreTest.scala 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..1adec18d1c --- /dev/null +++ b/apps/common/src/main/resources/db/migration/canton-network/postgres/stable/V073__sync_operator_acs_store.sql @@ -0,0 +1,24 @@ +-- 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_tid_mtm_mtd + on sync_operator_acs_store (store_id, migration_id, 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/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/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..169a4c3a43 --- /dev/null +++ b/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/SyncOperatorApp.scala @@ -0,0 +1,260 @@ +// 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.store.db.DbAppStore +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. + * + * The operator side of Amulet-funded traffic on a dedicated synchronizer. Its participant is + * connected both to the decentralized synchronizer, where purchases are burned and recorded, and + * to the synchronizer it operates, whose sequencer it grants the purchased traffic on. + * + * 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] = { + val synchronizerId = SynchronizerId.tryFromString(config.synchronizer.synchronizerId) + // MVP serves a single sequencer; see SyncOperatorSynchronizerConfig. + val sequencerConfig = config.synchronizer.sequencers.headOption.getOrElse( + throw Status.INVALID_ARGUMENT + .withDescription("No sequencer configured for the dedicated synchronizer") + .asRuntimeException() + ) + + 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( + sequencerConfig.adminApi, + appParameters.loggingConfig.api, + loggerFactory, + metrics.grpcClientMetrics, + retryProvider, + ) + _ <- appInitStep(s"Wait for the sequencer serving $synchronizerId") { + waitForSequencerServing(sequencerAdminConnection, synchronizerId) + } + // Resolved only because the store partitions its ingestion offsets by it. Purchases for a + // registered synchronizer are pinned to migration id 0 and are ingested regardless of it, + // see SyncOperatorStore.contractFilter. + domainMigrationId <- appInitStep(s"Resolving domain migration id") { + resolveDomainMigrationId(scanConnection) + } + storeKey = SyncOperatorStore.Key( + operatorParty = partyId, + dsoParty = dsoParty, + synchronizerId = synchronizerId, + ) + store = SyncOperatorStore( + storeKey, + storage, + loggerFactory, + retryProvider, + domainMigrationId, + participantId, + config.automation.ingestion, + config.parameters.defaultLimit, + ) + globalSynchronizerId <- appInitStep("Get the decentralized 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, + ) + } + } + + /** Fails initialization if the configured sequencer does not serve the configured synchronizer, + * rather than leaving a miswired node to discover that one purchase at a time. + */ + private def waitForSequencerServing( + sequencerAdminConnection: SequencerAdminConnection, + synchronizerId: SynchronizerId, + )(implicit traceContext: TraceContext): Future[Unit] = + retryProvider.waitUntil( + RetryFor.WaitingOnInitDependency, + "sync_operator_sequencer_serves_synchronizer", + s"the configured sequencer serves $synchronizerId", + sequencerAdminConnection.getStatus.map { status => + status.successOption.map(_.synchronizerId.logical) match { + case Some(`synchronizerId`) => () + case Some(served) => + throw Status.FAILED_PRECONDITION + .withDescription( + s"The configured sequencer serves $served, but this node is configured for $synchronizerId" + ) + .asRuntimeException() + case None => + throw Status.UNAVAILABLE + .withDescription("Sequencer is not yet initialized") + .asRuntimeException() + } + }, + logger, + ) + + private def resolveDomainMigrationId( + scanConnection: ScanConnection + )(implicit traceContext: TraceContext): Future[Long] = + DbAppStore.getHighestKnownMigrationId(storage).flatMap { + case Some(migrationId) => + logger.info(s"Resolved domain migration id $migrationId from the local store offsets") + Future.successful(migrationId) + case None => + retryProvider.getValueWithRetries( + RetryFor.WaitingOnInitDependency, + "sync_operator_domain_migration_id", + s"Wait for domain migration id to be available", + scanConnection.getMigrationId().map { migrationId => + logger.info(s"Resolved domain migration id $migrationId from scan") + migrationId + }, + 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..d68982cddd --- /dev/null +++ b/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/SyncOperatorAppBootstrap.scala @@ -0,0 +1,138 @@ +// 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] = { + // The node has no HTTP surface of its own yet: it is driven by on-ledger ingestion and the + // sequencer admin API. Operator-facing endpoints attach here when they land. + 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..4848dfd931 --- /dev/null +++ b/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/automation/SyncOperatorAutomationService.scala @@ -0,0 +1,73 @@ +// 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. + * + * The traffic reconciliation trigger, which turns the ingested purchases into sequencer traffic + * limits, is registered here. + */ +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 yet. The operator's participant is connected + // to the dedicated synchronizer, so it can be sourced from there when something does. + 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..a2ac2a3da8 --- /dev/null +++ b/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/config/SyncOperatorAppConfig.scala @@ -0,0 +1,60 @@ +// 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 + +case class SyncOperatorSequencerConfig( + adminApi: ClientConfig +) + +// The sequencers of the synchronizer this node serves. A single entry is the expected +// configuration; the field is a sequence so that a BFT synchronizer, which needs one reconciler +// per sequencer, does not require a config migration. +case class SyncOperatorSynchronizerConfig( + synchronizerId: String, + sequencers: Seq[SyncOperatorSequencerConfig], +) { + require(sequencers.nonEmpty, "at least one sequencer must be configured") +} + +case class SyncOperatorAppBackendConfig( + override val adminApi: AdminServerConfig = AdminServerConfig(), + override val storage: DbConfig, + postgres: SplicePostgresConfig = SplicePostgresConfig(), + // Ledger API user of the operator party, on a participant connected to both the decentralized + // and the dedicated synchronizer. + operatorUser: String, + participantClient: ParticipantClientConfig, + scanClient: ScanAppClientConfig, + synchronizer: SyncOperatorSynchronizerConfig, + override val automation: AutomationConfig = AutomationConfig(), + parameters: SpliceParametersConfig = SpliceParametersConfig(batching = BatchingConfig()), + trafficBalanceReconciliationDelay: NonNegativeFiniteDuration = + NonNegativeFiniteDuration.ofSeconds(10), + // 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..0b68699c85 --- /dev/null +++ b/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/store/SyncOperatorStore.scala @@ -0,0 +1,125 @@ +// 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. The buy choice sets + * the registered operator as an observer on each of them, so they are picked up on-ledger rather + * than by polling Scan. + */ +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 party registered as the operator of [[synchronizerId]], and the observer the buy choice + * sets on every purchase made for it. + */ + 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( + // Not filtered by this node's own migration id, unlike the SV store. A registered + // synchronizer is pinned to migration id 0 (upgrades go through LSU), so matching against + // the decentralized synchronizer's migration id would drop every purchase for this + // synchronizer the first time that synchronizer migrates. + mkFilter(splice.decentralizedsynchronizer.MemberTraffic.COMPANION)(co => + co.payload.dso == dso && + co.payload.operator.toScala.contains(operator) && + co.payload.synchronizerId == synchronizerId && + co.payload.migrationId == 0L + ) { 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 above has already established that these agree, so we take the parsed + // id from the key rather than parsing the payload again 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..5bd9a8c220 --- /dev/null +++ b/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/store/db/DbSyncOperatorStore.scala @@ -0,0 +1,109 @@ +// 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.codegen.java.splice.decentralizedsynchronizer.MemberTraffic +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, + 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.{QualifiedName, TemplateJsonDecoder} +import slick.jdbc.canton.ActionBasedSQLInterpolation.Implicits.actionBasedSQLInterpolationCanton + +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 SyncOperatorStore { + + override def dsoPartyId: PartyId = key.dsoParty + + override lazy val acsContractFilter: MultiDomainAcsStore.ContractFilter[ + SyncOperatorAcsStoreRowData, + AcsInterfaceViewRowData.NoInterfacesIngested, + ] = SyncOperatorStore.contractFilter(key) + + import multiDomainAcsStore.waitUntilAcsIngested + import org.lfdecentralizedtrust.splice.util.FutureUnlessShutdownUtil.futureUnlessShutdownToFuture + + private def acsStoreId: AcsStoreId = multiDomainAcsStore.acsStoreId + + override def getTotalPurchasedMemberTraffic(memberId: Member)(implicit + tc: TraceContext + ): Future[Long] = waitUntilAcsIngested { + for { + sum <- storage + .querySingle( + sql""" + select sum(total_traffic_purchased) + from #${SyncOperatorTables.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 = ${key.synchronizerId} + """.as[Long].headOption, + "getTotalPurchasedMemberTraffic", + ) + .value + } yield sum.getOrElse(0L) + } +} 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..d36b44d0ff --- /dev/null +++ b/apps/syncoperator/src/test/scala/org/lfdecentralizedtrust/splice/store/db/SyncOperatorStoreTest.scala @@ -0,0 +1,177 @@ +// 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) => + toActiveContract(dummyDomain, c, i.toLong) + }, + 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() + 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..e33c2847c4 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,18 @@ 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( + libraryDependencies ++= Seq(scalapb_runtime_grpc, scalapb_runtime), + BuildCommon.sharedAppSettings, + ) + lazy val pulumi = project .in(file("cluster/pulumi")) From 75cbcb9b94d0a1ad4dccd589af32035b3591080d Mon Sep 17 00:00:00 2001 From: sadiq1971 Date: Tue, 25 Aug 2026 23:51:09 +0600 Subject: [PATCH 2/5] fix: sync operator review fixes [ci] Drop the synchronizer id and sequencer list from the config and take the synchronizer id from the sequencer instead. Share the MemberTraffic sum query with the DSO store. Also fixes the store test, which never ran: the operator was missing as an observer on the ingested contracts, and the suite was absent from the non-integration test list. Signed-off-by: sadiq1971 --- .../store/db/MemberTrafficQueries.scala | 51 ++++++++++++++++ .../splice/store/StoreTestBase.scala | 3 +- .../splice/sv/store/db/DbSvDsoStore.scala | 29 ++++----- .../splice/syncoperator/SyncOperatorApp.scala | 59 +++++++------------ .../SyncOperatorAppBootstrap.scala | 3 +- .../SyncOperatorAutomationService.scala | 9 +-- .../config/SyncOperatorAppConfig.scala | 18 ++---- .../store/SyncOperatorStore.scala | 12 ++-- .../store/db/DbSyncOperatorStore.scala | 34 ++++------- .../store/db/SyncOperatorStoreTest.scala | 6 +- test-full-class-names-non-integration.log | 1 + 11 files changed, 111 insertions(+), 114 deletions(-) create mode 100644 apps/common/src/main/scala/org/lfdecentralizedtrust/splice/store/db/MemberTrafficQueries.scala 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/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 index 169a4c3a43..2478c861ad 100644 --- a/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/SyncOperatorApp.scala +++ b/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/SyncOperatorApp.scala @@ -38,9 +38,8 @@ import scala.concurrent.{ExecutionContextExecutor, Future} /** Class representing a sync operator app instance. * - * The operator side of Amulet-funded traffic on a dedicated synchronizer. Its participant is - * connected both to the decentralized synchronizer, where purchases are burned and recorded, and - * to the synchronizer it operates, whose sequencer it grants the purchased traffic on. + * Ingests the traffic purchases made for the synchronizer it operates and grants them on that + * synchronizer's sequencer. * * Modelled after Canton's ParticipantNode class. */ @@ -81,14 +80,6 @@ class SyncOperatorApp( partyId: PartyId, preInitializeState: Unit, )(implicit traceContext: TraceContext): Future[SyncOperatorApp.State] = { - val synchronizerId = SynchronizerId.tryFromString(config.synchronizer.synchronizerId) - // MVP serves a single sequencer; see SyncOperatorSynchronizerConfig. - val sequencerConfig = config.synchronizer.sequencers.headOption.getOrElse( - throw Status.INVALID_ARGUMENT - .withDescription("No sequencer configured for the dedicated synchronizer") - .asRuntimeException() - ) - for { scanConnection <- appInitStep(s"Get scan connection") { ScanConnection.singleCached( @@ -112,18 +103,17 @@ class SyncOperatorApp( } dsoParty <- appInitStep("Get DSO party id") { scanConnection.getDsoPartyId() } sequencerAdminConnection = new SequencerAdminConnection( - sequencerConfig.adminApi, + config.sequencer.adminApi, appParameters.loggingConfig.api, loggerFactory, metrics.grpcClientMetrics, retryProvider, ) - _ <- appInitStep(s"Wait for the sequencer serving $synchronizerId") { - waitForSequencerServing(sequencerAdminConnection, synchronizerId) + synchronizerId <- appInitStep("Get the synchronizer id from the sequencer") { + servedSynchronizerId(sequencerAdminConnection) } - // Resolved only because the store partitions its ingestion offsets by it. Purchases for a - // registered synchronizer are pinned to migration id 0 and are ingested regardless of it, - // see SyncOperatorStore.contractFilter. + // Only used to partition the store's ingestion offsets; purchases are ingested regardless + // of it, see SyncOperatorStore.contractFilter. domainMigrationId <- appInitStep(s"Resolving domain migration id") { resolveDomainMigrationId(scanConnection) } @@ -180,32 +170,23 @@ class SyncOperatorApp( } } - /** Fails initialization if the configured sequencer does not serve the configured synchronizer, - * rather than leaving a miswired node to discover that one purchase at a time. - */ - private def waitForSequencerServing( - sequencerAdminConnection: SequencerAdminConnection, - synchronizerId: SynchronizerId, - )(implicit traceContext: TraceContext): Future[Unit] = - retryProvider.waitUntil( + /** 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_sequencer_serves_synchronizer", - s"the configured sequencer serves $synchronizerId", - sequencerAdminConnection.getStatus.map { status => - status.successOption.map(_.synchronizerId.logical) match { - case Some(`synchronizerId`) => () - case Some(served) => - throw Status.FAILED_PRECONDITION - .withDescription( - s"The configured sequencer serves $served, but this node is configured for $synchronizerId" - ) - .asRuntimeException() - case None => + "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, ) 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 index d68982cddd..53d5ba2ee9 100644 --- a/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/SyncOperatorAppBootstrap.scala +++ b/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/SyncOperatorAppBootstrap.scala @@ -65,8 +65,7 @@ class SyncOperatorAppBootstrap( ) { override def initialize(adminRoutes: AdminRoutes): EitherT[Future, String, Unit] = { - // The node has no HTTP surface of its own yet: it is driven by on-ledger ingestion and the - // sequencer admin API. Operator-facing endpoints attach here when they land. + // No HTTP surface yet; operator-facing endpoints attach here. val _ = adminRoutes startInstanceUnlessClosing { new SyncOperatorApp( 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 index 4848dfd931..e606bac1f3 100644 --- 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 @@ -24,11 +24,7 @@ import org.lfdecentralizedtrust.splice.syncoperator.store.SyncOperatorStore import scala.concurrent.ExecutionContextExecutor -/** Manages background automation that runs on a sync operator app. - * - * The traffic reconciliation trigger, which turns the ingested purchases into sequencer traffic - * limits, is registered here. - */ +/** Manages background automation that runs on a sync operator app. */ class SyncOperatorAutomationService( automationConfig: AutomationConfig, clock: Clock, @@ -46,8 +42,7 @@ class SyncOperatorAutomationService( ) extends SpliceAppAutomationService( automationConfig, clock, - // Nothing registered here depends on domain time yet. The operator's participant is connected - // to the dedicated synchronizer, so it can be sourced from there when something does. + // Nothing registered here depends on domain time yet. DomainTimeSynchronization.Noop, store, ledgerClient, 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 index a2ac2a3da8..606dc8210d 100644 --- 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 @@ -15,30 +15,20 @@ import org.lfdecentralizedtrust.splice.config.{ } import org.lfdecentralizedtrust.splice.scan.config.ScanAppClientConfig +// The sequencer this node grants traffic on. case class SyncOperatorSequencerConfig( - adminApi: ClientConfig + adminApi: FullClientConfig ) -// The sequencers of the synchronizer this node serves. A single entry is the expected -// configuration; the field is a sequence so that a BFT synchronizer, which needs one reconciler -// per sequencer, does not require a config migration. -case class SyncOperatorSynchronizerConfig( - synchronizerId: String, - sequencers: Seq[SyncOperatorSequencerConfig], -) { - require(sequencers.nonEmpty, "at least one sequencer must be configured") -} - case class SyncOperatorAppBackendConfig( override val adminApi: AdminServerConfig = AdminServerConfig(), override val storage: DbConfig, postgres: SplicePostgresConfig = SplicePostgresConfig(), - // Ledger API user of the operator party, on a participant connected to both the decentralized - // and the dedicated synchronizer. + // Ledger API user of the operator party. operatorUser: String, participantClient: ParticipantClientConfig, scanClient: ScanAppClientConfig, - synchronizer: SyncOperatorSynchronizerConfig, + sequencer: SyncOperatorSequencerConfig, override val automation: AutomationConfig = AutomationConfig(), parameters: SpliceParametersConfig = SpliceParametersConfig(batching = BatchingConfig()), trafficBalanceReconciliationDelay: NonNegativeFiniteDuration = 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 index 0b68699c85..e41f5557bb 100644 --- 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 @@ -23,9 +23,8 @@ import scala.jdk.OptionConverters.* /** Store of a sync operator app. * - * Ingests the `MemberTraffic` purchases made for this operator's synchronizer. The buy choice sets - * the registered operator as an observer on each of them, so they are picked up on-ledger rather - * than by polling Scan. + * Ingests the `MemberTraffic` purchases made for this operator's synchronizer, which the buy choice + * makes it an observer of. */ trait SyncOperatorStore extends AppStore { @@ -68,9 +67,7 @@ object SyncOperatorStore { ) case class Key( - /** The party registered as the operator of [[synchronizerId]], and the observer the buy choice - * sets on every purchase made for it. - */ + /** The registered operator of [[synchronizerId]]. */ operatorParty: PartyId, /** The DSO party, sole signatory of `MemberTraffic`. */ dsoParty: PartyId, @@ -113,8 +110,7 @@ object SyncOperatorStore { // 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 above has already established that these agree, so we take the parsed - // id from the key rather than parsing the payload again per contract + // 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 index 5bd9a8c220..8c8f126cb7 100644 --- 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 @@ -8,7 +8,6 @@ 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.codegen.java.splice.decentralizedsynchronizer.MemberTraffic import org.lfdecentralizedtrust.splice.config.IngestionConfig import org.lfdecentralizedtrust.splice.environment.RetryProvider import org.lfdecentralizedtrust.splice.store.db.AcsQueries.AcsStoreId @@ -17,13 +16,13 @@ import org.lfdecentralizedtrust.splice.store.db.{ 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.{QualifiedName, TemplateJsonDecoder} -import slick.jdbc.canton.ActionBasedSQLInterpolation.Implicits.actionBasedSQLInterpolationCanton +import org.lfdecentralizedtrust.splice.util.TemplateJsonDecoder import scala.concurrent.{ExecutionContext, Future} @@ -69,6 +68,7 @@ class DbSyncOperatorStore( with AcsTables with AcsQueries with LimitHelpers + with MemberTrafficQueries with SyncOperatorStore { override def dsoPartyId: PartyId = key.dsoParty @@ -79,31 +79,19 @@ class DbSyncOperatorStore( ] = SyncOperatorStore.contractFilter(key) import multiDomainAcsStore.waitUntilAcsIngested - import org.lfdecentralizedtrust.splice.util.FutureUnlessShutdownUtil.futureUnlessShutdownToFuture private def acsStoreId: AcsStoreId = multiDomainAcsStore.acsStoreId override def getTotalPurchasedMemberTraffic(memberId: Member)(implicit tc: TraceContext ): Future[Long] = waitUntilAcsIngested { - for { - sum <- storage - .querySingle( - sql""" - select sum(total_traffic_purchased) - from #${SyncOperatorTables.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 = ${key.synchronizerId} - """.as[Long].headOption, - "getTotalPurchasedMemberTraffic", - ) - .value - } yield sum.getOrElse(0L) + sumPurchasedMemberTraffic( + storage, + SyncOperatorTables.acsTableName, + acsStoreId, + domainMigrationId, + memberId, + key.synchronizerId, + ) } } 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 index d36b44d0ff..3e2e038659 100644 --- 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 @@ -69,7 +69,9 @@ abstract class SyncOperatorStoreTest extends StoreTestBase with HasExecutionCont // the purchase is made and recorded on the decentralized synchronizer, not on the dedicated // one it names contracts.zipWithIndex.map { case (c, i) => - toActiveContract(dummyDomain, c, i.toLong) + // 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, @@ -134,6 +136,8 @@ abstract class SyncOperatorStoreTest extends StoreTestBase with HasExecutionCont "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 } 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 From 8f80966c869af7b9d1c30ed1cab6dc323b6b7191 Mon Sep 17 00:00:00 2001 From: sadiq1971 Date: Wed, 26 Aug 2026 03:17:02 +0600 Subject: [PATCH 3/5] fix: review follow-ups on the sync operator store and build [ci] Add package_name to the sync_operator_acs_store index, matching the shape V049 rebuilt the dso and scan indexes into; the shared MemberTraffic query filters on it. Drop trafficBalanceReconciliationDelay, which nothing reads, and the scalapb runtime deps, which the module has no generated code for. Add the module to clean-splice. Signed-off-by: sadiq1971 --- .../postgres/stable/V073__sync_operator_acs_store.sql | 5 +++-- .../splice/syncoperator/config/SyncOperatorAppConfig.scala | 2 -- .../splice/syncoperator/store/SyncOperatorStore.scala | 6 +----- build.sbt | 1 - project/BuildCommon.scala | 1 + 5 files changed, 5 insertions(+), 10 deletions(-) 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 index 1adec18d1c..e6035df96f 100644 --- 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 @@ -19,6 +19,7 @@ create table sync_operator_acs_store( total_traffic_purchased bigint ); -create index sync_operator_acs_store_sid_mid_tid_mtm_mtd - on sync_operator_acs_store (store_id, migration_id, template_id_qualified_name, member_traffic_member, member_traffic_domain) +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/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/config/SyncOperatorAppConfig.scala b/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/config/SyncOperatorAppConfig.scala index 606dc8210d..0204f28a27 100644 --- 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 @@ -31,8 +31,6 @@ case class SyncOperatorAppBackendConfig( sequencer: SyncOperatorSequencerConfig, override val automation: AutomationConfig = AutomationConfig(), parameters: SpliceParametersConfig = SpliceParametersConfig(batching = BatchingConfig()), - trafficBalanceReconciliationDelay: NonNegativeFiniteDuration = - NonNegativeFiniteDuration.ofSeconds(10), // 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. 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 index e41f5557bb..6e0756994b 100644 --- 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 @@ -93,15 +93,11 @@ object SyncOperatorStore { MultiDomainAcsStore.SimpleContractFilter( key.operatorParty, Map( - // Not filtered by this node's own migration id, unlike the SV store. A registered - // synchronizer is pinned to migration id 0 (upgrades go through LSU), so matching against - // the decentralized synchronizer's migration id would drop every purchase for this - // synchronizer the first time that synchronizer migrates. mkFilter(splice.decentralizedsynchronizer.MemberTraffic.COMPANION)(co => co.payload.dso == dso && co.payload.operator.toScala.contains(operator) && co.payload.synchronizerId == synchronizerId && - co.payload.migrationId == 0L + co.payload.migrationId == 0L // A registered synchronizer is pinned to migration id 0. ) { contract => SyncOperatorAcsStoreRowData( contract, diff --git a/build.sbt b/build.sbt index e33c2847c4..5804ad1e0e 100644 --- a/build.sbt +++ b/build.sbt @@ -1906,7 +1906,6 @@ lazy val `apps-syncoperator` = `apps-scan` % "compile->compile;test->test", ) .settings( - libraryDependencies ++= Seq(scalapb_runtime_grpc, scalapb_runtime), BuildCommon.sharedAppSettings, ) 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", From bca0d2feed6457eb4df31574f7946af092ad15ad Mon Sep 17 00:00:00 2001 From: sadiq1971 Date: Thu, 27 Aug 2026 18:08:48 +0600 Subject: [PATCH 4/5] fix: review wording on the sync operator app [ci] Drop the speculative "yet" from the domain time comment, and call the init step what the variable already calls it. Signed-off-by: sadiq1971 --- .../splice/syncoperator/SyncOperatorApp.scala | 2 +- .../syncoperator/automation/SyncOperatorAutomationService.scala | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) 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 index 2478c861ad..58f39f47fa 100644 --- a/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/SyncOperatorApp.scala +++ b/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/SyncOperatorApp.scala @@ -132,7 +132,7 @@ class SyncOperatorApp( config.automation.ingestion, config.parameters.defaultLimit, ) - globalSynchronizerId <- appInitStep("Get the decentralized synchronizer id") { + globalSynchronizerId <- appInitStep("Get the global synchronizer id") { scanConnection.getAmuletRulesDomain()(traceContext) } readOnlyLedgerConnection = ledgerClient 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 index e606bac1f3..3821e1e148 100644 --- 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 @@ -42,7 +42,7 @@ class SyncOperatorAutomationService( ) extends SpliceAppAutomationService( automationConfig, clock, - // Nothing registered here depends on domain time yet. + // Nothing registered here depends on domain time. DomainTimeSynchronization.Noop, store, ledgerClient, From 6021ca72c6ca3b24748933b996b395c37c05f1eb Mon Sep 17 00:00:00 2001 From: sadiq1971 Date: Thu, 27 Aug 2026 19:58:31 +0600 Subject: [PATCH 5/5] fix: pin the sync operator store's migration id to 0 [ci] MIGRATION_ID is frozen network wide and logical synchronizer upgrades carry a serial id instead, so the store's partition can never move and resolving it bought nothing. Drops resolveDomainMigrationId and its scan fallback. It also lines the store's stamp up with the payload.migrationId == 0L check in the contract filter, so there is no longer a second migration id a few lines away meaning something different. Signed-off-by: sadiq1971 --- .../splice/syncoperator/SyncOperatorApp.scala | 30 ++----------------- 1 file changed, 3 insertions(+), 27 deletions(-) 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 index 58f39f47fa..30fcd6a1da 100644 --- a/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/SyncOperatorApp.scala +++ b/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/SyncOperatorApp.scala @@ -27,7 +27,6 @@ import org.lfdecentralizedtrust.splice.environment.{ SpliceLedgerClient, } import org.lfdecentralizedtrust.splice.scan.admin.api.client.ScanConnection -import org.lfdecentralizedtrust.splice.store.db.DbAppStore import org.lfdecentralizedtrust.splice.syncoperator.automation.SyncOperatorAutomationService import org.lfdecentralizedtrust.splice.syncoperator.config.SyncOperatorAppBackendConfig import org.lfdecentralizedtrust.splice.syncoperator.metrics.SyncOperatorAppMetrics @@ -112,11 +111,6 @@ class SyncOperatorApp( synchronizerId <- appInitStep("Get the synchronizer id from the sequencer") { servedSynchronizerId(sequencerAdminConnection) } - // Only used to partition the store's ingestion offsets; purchases are ingested regardless - // of it, see SyncOperatorStore.contractFilter. - domainMigrationId <- appInitStep(s"Resolving domain migration id") { - resolveDomainMigrationId(scanConnection) - } storeKey = SyncOperatorStore.Key( operatorParty = partyId, dsoParty = dsoParty, @@ -127,7 +121,9 @@ class SyncOperatorApp( storage, loggerFactory, retryProvider, - domainMigrationId, + // 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, @@ -190,26 +186,6 @@ class SyncOperatorApp( logger, ) - private def resolveDomainMigrationId( - scanConnection: ScanConnection - )(implicit traceContext: TraceContext): Future[Long] = - DbAppStore.getHighestKnownMigrationId(storage).flatMap { - case Some(migrationId) => - logger.info(s"Resolved domain migration id $migrationId from the local store offsets") - Future.successful(migrationId) - case None => - retryProvider.getValueWithRetries( - RetryFor.WaitingOnInitDependency, - "sync_operator_domain_migration_id", - s"Wait for domain migration id to be available", - scanConnection.getMigrationId().map { migrationId => - logger.info(s"Resolved domain migration id $migrationId from scan") - migrationId - }, - logger, - ) - } - protected[this] override def automationServices(st: SyncOperatorApp.State) = Seq(st.automation) }