feat: sync operator app skeleton

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;

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)
+}

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)
 }

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,
     )
 

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(

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)
+  }
+}