feat: let the sync operator set an outage traffic allowance

apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/EnvironmentDefinition.scala

@@ -523,22 +523,26 @@ case class EnvironmentDefinition(
   /** Drops the splitwell apps and leaves alice's validator as the only one connected to the
     * splitwell synchronizer.
     */
-  def withOnlyAliceValidatorConnectingToSplitwell: EnvironmentDefinition = {
+  def withOnlyAliceValidatorConnectingToSplitwell: EnvironmentDefinition =
+    withValidatorsDisconnectedFromSplitwell("bobValidator", "splitwellValidator")
+
+  /** Drops all splitwell apps, and clears all of the given validators' extra synchronizers, not
+    * just splitwell, so the global synchronizer is the only one they connect to.
+    */
+  def withValidatorsDisconnectedFromSplitwell(validators: String*): EnvironmentDefinition =
     this
       .addConfigTransform((_, conf) =>
         conf.copy(
           splitwellApps = Map.empty,
-          validatorApps =
-            Seq("bobValidator", "splitwellValidator").foldLeft(conf.validatorApps) { (apps, name) =>
-              apps.updatedWith(InstanceName.tryCreate(name)) {
-                _.map { validatorConfig =>
-                  validatorConfig.copy(domains = validatorConfig.domains.copy(extra = Seq.empty))
-                }
+          validatorApps = validators.foldLeft(conf.validatorApps) { (apps, name) =>
+            apps.updatedWith(InstanceName.tryCreate(name)) {
+              _.map { validatorConfig =>
+                validatorConfig.copy(domains = validatorConfig.domains.copy(extra = Seq.empty))
               }
-            },
+            }
+          },
         )
       )
-  }
 
   def clearConfigTransforms(): EnvironmentDefinition =
     copy(configTransformsWithContext = _ => Seq())

apps/app/src/test/scala/org/lfdecentralizedtrust/splice/util/SyncOperatorTestUtil.scala

@@ -9,6 +9,7 @@ import org.lfdecentralizedtrust.splice.codegen.java.splice
 import org.lfdecentralizedtrust.splice.codegen.java.splice.decentralizedsynchronizer.RegisteredSynchronizer
 import org.lfdecentralizedtrust.splice.codegen.java.splice.round.IssuingMiningRound
 import org.lfdecentralizedtrust.splice.codegen.java.splice.types.Round
+import org.lfdecentralizedtrust.splice.console.SyncOperatorAppBackendReference
 import org.lfdecentralizedtrust.splice.environment.SequencerAdminConnection
 import org.lfdecentralizedtrust.splice.integration.tests.SpliceTests.{
   SpliceTestConsoleEnvironment,
@@ -82,10 +83,18 @@ trait SyncOperatorTestUtil extends TestCommon {
       .exerciseResult
   }
 
+  /** The sync operator app that is running, as a test may run another instance in its place. */
+  protected def runningSyncOperatorBackend(implicit
+      env: SpliceTestConsoleEnvironment
+  ): SyncOperatorAppBackendReference =
+    env.syncOperators.local
+      .find(_.is_running)
+      .getOrElse(fail("No sync operator app is running"))
+
   protected def trafficState(
       member: Member
   )(implicit env: SpliceTestConsoleEnvironment): Option[SequencerAdminConnection.TrafficState] =
-    syncOperatorBackend.appState.sequencerAdminConnection
+    runningSyncOperatorBackend.appState.sequencerAdminConnection
       .lookupSequencerTrafficControlState(member)
       .futureValue
 

apps/common/src/main/scala/org/lfdecentralizedtrust/splice/automation/ReconcileSequencerLimitWithMemberTrafficTriggerBase.scala

@@ -19,10 +19,11 @@ import org.lfdecentralizedtrust.splice.util.AssignedContract
 import scala.concurrent.{ExecutionContext, Future}
 
 /** Grants the traffic purchased via `MemberTraffic` contracts on the sequencer this trigger's
-  * connection serves, skipping contracts that name another synchronizer. One instance per
-  * sequencer, so a BFT synchronizer runs one operator app per sequencer.
+  * connection serves, skipping contracts that name another synchronizer. Only the SV app's
+  * `ReconcileSequencerLimitWithMemberTrafficTrigger` extends it: each SV runs one, which grants on
+  * that SV's own sequencer of the decentralized synchronizer.
   *
-  * This trigger currently relies on enough nodes working on the same set traffic balance request
+  * This trigger currently relies on enough SVs working on the same set traffic balance request
   * around the same time. It also depends on the sorting of tasks done in OnAssignedContractTrigger
   * to make this more likely to succeed.
   *

apps/common/src/main/scala/org/lfdecentralizedtrust/splice/store/db/MemberTrafficQueries.scala

@@ -48,4 +48,41 @@ trait MemberTrafficQueries extends AcsJdbcTypes with LimitHelpers { this: NamedL
         )
         .value
     } yield sum.getOrElse(0L)
+
+  /** Total traffic purchased on `synchronizerId`, for every member with a purchase on record. */
+  protected def sumPurchasedMemberTrafficPerMember(
+      storage: DbStorage,
+      acsTableName: String,
+      acsStoreId: AcsStoreId,
+      migrationId: Long,
+      synchronizerId: SynchronizerId,
+  )(implicit
+      ec: ExecutionContext,
+      tc: TraceContext,
+      closeContext: CloseContext,
+  ): Future[Map[Member, Long]] = {
+    // Throws on a member that does not parse: ingestion stores a member only once it parses, so
+    // that would be a corrupt row.
+    import Member.DbStorageImplicits.getResultMember
+    for {
+      rows <- storage
+        .query(
+          sql"""
+               select member_traffic_member, 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_domain = $synchronizerId
+                and member_traffic_member is not null
+               group by member_traffic_member
+             """.as[(Member, Long)],
+          // the callers' method name, which is what the DB retry logs report
+          "listTotalPurchasedMemberTraffic",
+        )
+    } yield rows.toMap
+  }
 }

apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/SyncOperatorApp.scala

@@ -173,6 +173,7 @@ class SyncOperatorApp(
         config.lsu,
         config.lsuDumpPath,
         config.trafficBalanceReconciliationDelay,
+        config.outageTrafficAllowance,
         loggerFactory,
         packageVersionSupport,
       )

apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/automation/ReconcileDedicatedSequencerTrafficTrigger.scala

@@ -3,50 +3,166 @@
 
 package org.lfdecentralizedtrust.splice.syncoperator.automation
 
+import cats.syntax.apply.*
 import com.digitalasset.canton.config.NonNegativeFiniteDuration
-import com.digitalasset.canton.topology.Member
+import com.digitalasset.canton.config.RequireTypes.{NonNegativeLong, PositiveLong}
+import com.digitalasset.canton.logging.pretty.{Pretty, PrettyPrinting}
+import com.digitalasset.canton.topology.{Member, ParticipantId}
 import com.digitalasset.canton.tracing.TraceContext
 import io.opentelemetry.api.trace.Tracer
 import org.apache.pekko.stream.Materializer
 import org.lfdecentralizedtrust.splice.automation.{
-  ReconcileSequencerLimitWithMemberTrafficTriggerBase,
+  PollingParallelTaskExecutionTrigger,
+  TaskOutcome,
+  TaskSuccess,
   TriggerContext,
 }
-import org.lfdecentralizedtrust.splice.environment.SequencerAdminConnection
-import org.lfdecentralizedtrust.splice.syncoperator.store.SyncOperatorStore
 import org.lfdecentralizedtrust.splice.environment.SynchronizerNodeService
+import org.lfdecentralizedtrust.splice.environment.TopologyAdminConnection.TopologySnapshot
 import org.lfdecentralizedtrust.splice.syncoperator.SyncOperatorSynchronizerNode
+import org.lfdecentralizedtrust.splice.syncoperator.automation.ReconcileDedicatedSequencerTrafficTrigger.Task
+import org.lfdecentralizedtrust.splice.syncoperator.store.SyncOperatorStore
 
 import scala.concurrent.{ExecutionContext, Future}
 
-/** Reconciles the traffic purchased for this operator's synchronizer with its sequencer */
+/** Keeps the traffic limit of every participant on this operator's sequencer at the traffic
+  * purchased for it on this synchronizer, plus the outage traffic allowance while that is set.
+  *
+  * Each poll reads the traffic state of every member the sequencer knows and the purchased total of
+  * every member, so a participant that joins later, also during an outage, gets its limit on the
+  * next poll. A participant with no purchase on record has bought nothing, so it gets the allowance
+  * alone. Mediators are left to [[MediatorUnlimitedTrafficTrigger]], and sequencers are not charged
+  * traffic.
+  *
+  * Limits are set exactly, both ways: setting the allowance raises them, and removing it lowers
+  * them back to the purchased totals, so a participant that used the allowance is below zero until
+  * its purchases cover what it consumed.
+  *
+  * Reading every member on each poll is fine for the few members of a dedicated synchronizer, but
+  * would not scale to the global synchronizer.
+  *
+  * On a synchronizer whose sequencers several operators run, a limit changes only once the
+  * sequencer group's threshold of its sequencers have sent the same one, so the operators have to
+  * set and remove the allowance together. Whenever their targets differ, as while only some have
+  * the allowance set or while one operator's store has not ingested a purchase yet, limits only
+  * stay stable if that threshold is above half the active sequencers: with a lower one, the
+  * sequencers on each side can reach it and override each other's limits on every poll. Above
+  * half, a split where neither side reaches it leaves the limits where they are until enough of
+  * them send the same limit.
+  */
 class ReconcileDedicatedSequencerTrafficTrigger(
     override protected val context: TriggerContext,
     store: SyncOperatorStore,
     synchronizerNodeService: SynchronizerNodeService[SyncOperatorSynchronizerNode],
     trafficBalanceReconciliationDelay: NonNegativeFiniteDuration,
+    outageTrafficAllowance: Option[PositiveLong],
 )(implicit
-    ec: ExecutionContext,
+    override val ec: ExecutionContext,
     mat: Materializer,
-    tracer: Tracer,
-) extends ReconcileSequencerLimitWithMemberTrafficTriggerBase(
-      store,
-      trafficBalanceReconciliationDelay,
-    ) {
-
-  override protected def sequencerAdminConnection()(implicit
-      tc: TraceContext
-  ): Future[SequencerAdminConnection] =
-    synchronizerNodeService.sequencerAdminConnection()
-
-  override protected def getTotalPurchasedMemberTraffic(memberId: Member)(implicit
-      tc: TraceContext
-  ): Future[Long] =
-    store.getTotalPurchasedMemberTraffic(memberId)
-
-  override protected def trafficLimitOffset(memberId: Member)(implicit
-      tc: TraceContext
-  ): Future[Either[String, Long]] =
-    // No prior consumption to carry.
-    Future.successful(Right(0L))
+    override val tracer: Tracer,
+) extends PollingParallelTaskExecutionTrigger[Task] {
+
+  private val allowance = outageTrafficAllowance.map(_.value)
+
+  override protected def retrieveTasks()(implicit tc: TraceContext): Future[Seq[Task]] =
+    for {
+      connection <- synchronizerNodeService.sequencerAdminConnection()
+      (states, totals) <- (
+        // An empty filter lists every member the sequencer knows.
+        connection.listSequencerTrafficControlState(),
+        store.listTotalPurchasedMemberTraffic(),
+      ).tupled
+    } yield ReconcileDedicatedSequencerTrafficTrigger.tasks(
+      states.map(state => state.member -> state.extraTrafficLimit.value).toMap,
+      totals,
+      allowance,
+    )
+
+  override protected def completeTask(task: Task)(implicit tc: TraceContext): Future[TaskOutcome] =
+    for {
+      connection <- synchronizerNodeService.sequencerAdminConnection()
+      // Read again, so the grant carries the serial the sequencer expects next.
+      (trafficState, sequencerState) <- (
+        connection.getSequencerTrafficControlState(task.member),
+        connection.getSequencerSynchronizerState(TopologySnapshot.Effective),
+      ).tupled
+      // And the total after it: a purchase granted since the task was retrieved is then either in
+      // the total, or its grant moved the serial and this one is refused and retried.
+      current <- currentTask(task)
+      limit = trafficState.extraTrafficLimit.value
+      _ <- connection.setSequencerTrafficControlState(
+        trafficState,
+        sequencerState,
+        NonNegativeLong.tryCreate(current.target),
+        context.pollingClock,
+        trafficBalanceReconciliationDelay,
+      )
+    } yield TaskSuccess(s"Set the traffic limit of ${task.member} from $limit to ${current.target}")
+
+  override protected def isStaleTask(task: Task)(implicit tc: TraceContext): Future[Boolean] =
+    for {
+      connection <- synchronizerNodeService.sequencerAdminConnection()
+      state <- connection.lookupSequencerTrafficControlState(task.member)
+      current <- currentTask(task)
+    } yield state.forall(state => current.isDone(state.extraTrafficLimit.value))
+
+  // The task with the member's purchased total as it is now, so a purchase ingested since the task
+  // was retrieved is not taken back.
+  private def currentTask(task: Task)(implicit tc: TraceContext): Future[Task] =
+    store.getTotalPurchasedMemberTraffic(task.member).map(total => task.copy(total = total))
+}
+
+object ReconcileDedicatedSequencerTrafficTrigger {
+
+  /** Whether this trigger sets the member's limit. Only participants: a mediator has no purchase
+    * path and is granted unlimited traffic by [[MediatorUnlimitedTrafficTrigger]], which a purchase
+    * someone made for it must not lower, and a sequencer is not charged traffic.
+    */
+  private[automation] def isCovered(member: Member): Boolean = member match {
+    case _: ParticipantId => true
+    case _ => false
+  }
+
+  /** The grants that bring each participant on the sequencer to its target, in a fixed order.
+    *
+    * @param limits
+    *   the current limit of every member the sequencer has a traffic state for
+    * @param totals
+    *   the purchased total of every member with a purchase on record for this synchronizer. A
+    *   purchase for a member without a traffic state waits until the sequencer has one.
+    */
+  def tasks(
+      limits: Map[Member, Long],
+      totals: Map[Member, Long],
+      allowance: Option[Long],
+  ): Seq[Task] =
+    limits.toSeq
+      .filter { case (member, _) => isCovered(member) }
+      .sortBy { case (member, _) => member.toProtoPrimitive }
+      .flatMap { case (member, limit) =>
+        val task = Task(member, totals.getOrElse(member, 0L), allowance)
+        Option.when(!task.isDone(limit))(task)
+      }
+
+  final case class Task(member: Member, total: Long, allowance: Option[Long])
+      extends PrettyPrinting {
+
+    /** The purchased total, plus the allowance while it is set, capped at the largest limit there
+      * is.
+      */
+    def target: Long =
+      allowance.fold(total)(a => if (a > Long.MaxValue - total) Long.MaxValue else total + a)
+
+    /** Whether `limit` leaves nothing to do. Exact both ways, so that removing the allowance lowers
+      * a limit it raised.
+      */
+    def isDone(limit: Long): Boolean = limit == target
+
+    override def pretty: Pretty[Task] = prettyOfClass(
+      param("member", _.member),
+      param("total", _.total),
+      paramIfDefined("allowance", _.allowance),
+      param("target", _.target),
+    )
+  }
 }

apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/automation/SyncOperatorAutomationService.scala

@@ -4,6 +4,7 @@
 package org.lfdecentralizedtrust.splice.syncoperator.automation
 
 import com.digitalasset.canton.config.NonNegativeFiniteDuration
+import com.digitalasset.canton.config.RequireTypes.PositiveLong
 import com.digitalasset.canton.logging.NamedLoggerFactory
 import com.digitalasset.canton.resource.DbStorage
 import com.digitalasset.canton.time.Clock
@@ -48,6 +49,7 @@ class SyncOperatorAutomationService(
     lsuConfig: Option[SyncOperatorLsuConfig],
     lsuDumpPath: Option[Path],
     trafficBalanceReconciliationDelay: NonNegativeFiniteDuration,
+    outageTrafficAllowance: Option[PositiveLong],
     protected val loggerFactory: NamedLoggerFactory,
     packageVersionSupport: PackageVersionSupport,
 )(implicit
@@ -99,6 +101,7 @@ class SyncOperatorAutomationService(
       store,
       synchronizerNodeService,
       trafficBalanceReconciliationDelay,
+      outageTrafficAllowance,
     )
   )
 

apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/config/SyncOperatorAppConfig.scala

@@ -9,7 +9,7 @@ import com.digitalasset.canton.admin.api.client.data.{
   SynchronizerLimits,
 }
 import com.digitalasset.canton.config.*
-import com.digitalasset.canton.config.RequireTypes.NonNegativeInt
+import com.digitalasset.canton.config.RequireTypes.{NonNegativeInt, PositiveLong}
 import com.digitalasset.canton.version.ProtocolVersion
 import org.lfdecentralizedtrust.splice.config.{
   AutomationConfig,
@@ -96,6 +96,17 @@ case class SyncOperatorAppBackendConfig(
     parameters: SpliceParametersConfig = SpliceParametersConfig(batching = BatchingConfig()),
     trafficBalanceReconciliationDelay: NonNegativeFiniteDuration =
       NonNegativeFiniteDuration.ofSeconds(10),
+    // Bytes each participant on this synchronizer may use past what has been bought for it. The
+    // operator sets it while the global synchronizer is unavailable, so nothing can be bought, and
+    // removes it once it is back. Setting or removing it only takes effect on a restart of this
+    // app, and the app only starts once the configured Scan is reachable. While it is set, this app
+    // keeps every participant's limit at its purchased total plus this; while it is unset, at its
+    // purchased total. Every participant gets it, including ones with no purchase on record and
+    // ones that join while it is set, so keep onboarding to the synchronizer restricted meanwhile.
+    // Where several operators run the synchronizer's sequencers, they set and remove it together.
+    // While their targets differ, as while only some have it set, limits only stay stable with a
+    // sequencer threshold above half the active sequencers.
+    outageTrafficAllowance: Option[PositiveLong] = None,
     // 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.