diff --git a/apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/tests/SyncOperatorTrafficIntegrationTest.scala b/apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/tests/SyncOperatorTrafficIntegrationTest.scala index 82f6f7a6fc3..d7d98445022 100644 --- a/apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/tests/SyncOperatorTrafficIntegrationTest.scala +++ b/apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/tests/SyncOperatorTrafficIntegrationTest.scala @@ -33,6 +33,7 @@ import org.lfdecentralizedtrust.splice.util.{ } import org.lfdecentralizedtrust.splice.validator.automation.TopupMemberTrafficTrigger import org.lfdecentralizedtrust.splice.wallet.store.TxLogEntry +import org.lfdecentralizedtrust.splice.wallet.util.ExtraTrafficTopupParameters import org.slf4j.event.Level import scala.concurrent.duration.* @@ -93,7 +94,22 @@ class SyncOperatorTrafficIntegrationTest ConfigTransforms.updateAllValidatorConfigs { case (name, validatorConfig) => if (name == "bobValidator") validatorConfig - // Paused throughout, so bob's traffic on splitwell is only what the test buys him. + .focus(_.domains.global.buyExtraTraffic.targetThroughput) + .replace(NonNegativeNumeric.tryCreate(BigDecimal(0))) + .focus(_.domains.extra) + .modify(_.map { extra => + if (extra.alias == splitwellAlias) + extra + // 100 bytes a second over a two-minute interval: less per top-up than the + // shortfall bob is left with once the allowance is removed, so one top-up only + // covers it if it buys the shortfall too. + .focus(_.topup.targetThroughput) + .replace(NonNegativeNumeric.tryCreate(BigDecimal(100))) + .focus(_.topup.minTopupInterval) + .replace(NonNegativeFiniteDuration.ofMinutes(2)) + else extra + }) + // Resumed only once the outage test needs bob's top-up to cover his shortfall. .focus(_.automation) .modify(_.withPausedTrigger[TopupMemberTrafficTrigger]) else if (name == "aliceValidator") @@ -525,6 +541,49 @@ class SyncOperatorTrafficIntegrationTest ) } + // Refused, bob consumes nothing more, so this is what his top-up finds. + val shortfall = consumed(bob) - totalsWithoutAllowance(bob) + // As bob's top-up trigger derives it from his splitwell target. + val bobTopupAmount = { + val topup = bobValidatorBackend.config.domains.extra + .find(_.alias == splitwellAlias) + .value + .topup + ExtraTrafficTopupParameters( + topup.targetThroughput, + topup.minTopupInterval, + sv1ScanBackend + .getAmuletConfigAsOf(env.environment.clock.now) + .decentralizedSynchronizer + .fees + .minTopupAmount, + bobValidatorBackend.config.automation.topupTriggerPollingInterval_, + ).topupAmount + } + clue("bob's shortfall is more than his top-up, so the top-up alone does not cover it") { + shortfall should be > bobTopupAmount + } + clue("bob's participant, whose view his top-up reads, sees the same shortfall") { + eventually() { + bobParticipant.traffic_control + .traffic_state(synchronizerId) + .extraTrafficRemainder shouldBe -shortfall + } + } + val topupTrigger = + bobValidatorBackend.appState.automation.trigger[TopupMemberTrafficTrigger] + setTriggersWithin(triggersToResumeAtStart = Seq(topupTrigger)) { + clue("one top-up buys bob's shortfall with it, and he transacts again") { + // Inside bob's two-minute top-up interval, so no second top-up is bought meanwhile. + eventually(100.seconds) { + extraTrafficLimit(bob) should be > consumed(bob) + } + // One purchase of the top-up and the shortfall together, not the top-up alone or two + // top-ups. + purchased()(bob) - totalsWithoutAllowance(bob) shouldBe bobTopupAmount + shortfall + ping(bobParticipant, synchronizerId) + } + } } } diff --git a/apps/validator/src/main/scala/org/lfdecentralizedtrust/splice/validator/automation/TopupMemberTrafficTrigger.scala b/apps/validator/src/main/scala/org/lfdecentralizedtrust/splice/validator/automation/TopupMemberTrafficTrigger.scala index d25083443a7..30c6ca4aafb 100644 --- a/apps/validator/src/main/scala/org/lfdecentralizedtrust/splice/validator/automation/TopupMemberTrafficTrigger.scala +++ b/apps/validator/src/main/scala/org/lfdecentralizedtrust/splice/validator/automation/TopupMemberTrafficTrigger.scala @@ -60,6 +60,7 @@ import org.apache.pekko.stream.Materializer import java.time.Instant import java.util.Optional +import scala.collection.concurrent.TrieMap import scala.concurrent.{ExecutionContext, Future} import scala.jdk.CollectionConverters.* import scala.jdk.OptionConverters.* @@ -82,6 +83,16 @@ class TopupMemberTrafficTrigger( private val validator = store.key.validatorParty + // Per target, the traffic state that its last shortfall purchase bought the shortfall of. The + // participant's traffic state only reflects a purchase once the synchronizer's operator has + // granted it, so until that state is updated, the shortfall it shows is already bought. Any + // traffic update ends that, also a grant of some other purchase while ours is still pending, + // and the remaining shortfall is then bought again: that traffic stays in the member's balance, + // so it is only bought early. Only a purchase seen to succeed is recorded, and only in memory, + // so after a restart, or after a shortfall purchase whose outcome was not seen, a shortfall whose + // purchase is not granted yet is also bought again. + private val lastShortfallPurchases = TrieMap.empty[SynchronizerId, TrafficState] + override protected def retrieveTasks()(implicit tc: TraceContext ): Future[Seq[TopupMemberTrafficTrigger.Task]] = { @@ -116,34 +127,69 @@ class TopupMemberTrafficTrigger( ) validatorWallet <- ValidatorUtil.getValidatorWallet(store, walletManager) budget <- TopupUtil.topupBudget(scanConnection, validatorWallet.store) - candidates <- MonadUtil.sequentialTraverseFilter(targets)(target => + candidates <- MonadUtil.sequentialTraverse(targets)(target => // The traversal is sequential, so a failure would otherwise cost every remaining target // its top-up for this poll. Info, not warn: a synchronizer is legitimately unreachable // for a stretch of polls while it is being upgraded or recovered. retrieveTaskFor(target, activeSynchronizerId).recover { case ex => logger.info(s"Skipping the top-up for ${target.alias} in this poll", ex) - None + Seq.empty } ) - (funded, unfunded) = TopupMemberTrafficTrigger.fundedTasks(candidates, budget) + (funded, unfunded) = TopupMemberTrafficTrigger.fundedTasks( + candidates.filter(_.nonEmpty), + budget, + )(_.unblocks) // we do not even submit the topup tx if the validator does not have sufficient funds because we know // the tx would fail but it would still drain synchronizer traffic which we would like to avoid (see #11915). - _ = unfunded.foreach(task => + _ = unfunded.foreach { task => + val topupAmount = task.target.topupParameters.topupAmount logger.warn( s"Insufficient funds to buy configured traffic amount. Please ensure that the validator's wallet has enough amulets to purchase " + - s"${BigDecimal(task.target.topupParameters.topupAmount) / 1e6} MB of traffic on ${task.target.alias} to continue healthy operation." + s"${BigDecimal(topupAmount) / 1e6} MB of traffic on ${task.target.alias} to continue healthy operation." + + (if ( + task.shortfall > 0 && !task.shortfallAwaitingGrant && + !TopupMemberTrafficTrigger.unblocks( + topupAmount, + task.trafficState.extraTrafficRemainder, + ) + ) + s" This member is also ${BigDecimal(task.shortfall) / 1e6} MB below zero there: for one purchase to get it submitting again, the wallet needs enough amulets for " + + s"${BigDecimal(topupAmount + task.shortfall) / 1e6} MB, the top-up and the shortfall together." + else "") ) - ) + } + // A member below zero that gets the top-up alone, because the wallet does not cover both. + _ = funded + .filter(task => + task.shortfall > 0 && !task.shortfallAwaitingGrant && task.extraTraffic == 0 + ) + .foreach { task => + val message = + s"Insufficient funds to buy the ${BigDecimal(task.shortfall) / 1e6} MB of traffic this member is below zero on ${task.target.alias} " + + "together with the configured top-up, so only the top-up will be bought" + // The top-up alone takes the member above zero, so it can submit all the same. + if (task.unblocks) logger.info(s"$message, which covers it.") + else + logger.warn( + s"$message, and the member cannot submit there until its top-ups have covered it. " + + "Please ensure that the validator's wallet has enough amulets to buy both." + ) + } } yield funded } - /** The task this target is due, paired with what it would cost the wallet. */ + /** The purchases this target is due, preferred first, each paired with what it would cost the + * wallet: for a member below zero, its shortfall together with the configured top-up, and the + * top-up alone in case the wallet does not cover both. Only the top-up alone while an earlier + * purchase of that shortfall is not granted yet. + */ private def retrieveTaskFor( target: TopupMemberTrafficTrigger.Target, submissionSynchronizerId: SynchronizerId, )(implicit tc: TraceContext - ): Future[Option[(TopupMemberTrafficTrigger.Task, BigDecimal)]] = + ): Future[Seq[(TopupMemberTrafficTrigger.Task, BigDecimal)]] = for { currentTrafficState <- participantAdminConnection.getParticipantTrafficState( target.synchronizerId @@ -151,44 +197,64 @@ class TopupMemberTrafficTrigger( registration <- if (!target.needsRegistration) Future.successful(None) else scanConnection.lookupSynchronizerRegistration(target.synchronizerId.toProtoPrimitive) - task <- + purchases <- if (target.needsRegistration && registration.isEmpty) { // A buy without it is rejected on-ledger, and lastPurchasedAt never advances, so // nothing would stop the retry. logger.info( s"Not topping up ${target.alias}: Scan serves no registration for ${target.synchronizerId}" ) - Future.successful(None) + Future.successful(Seq.empty) } else for { topupState <- getOrCreateValidatorTopupState(target, submissionSynchronizerId) - cost <- - if (!wantsTopup(target, currentTrafficState, topupState)) Future.successful(None) - else - TopupUtil - // The cost is derived from this target's throughput and interval. - .minWalletBalanceForTopup( - scanConnection, - target.topupConfig, - clock, - registration, + tasks <- + if (!wantsTopup(target, currentTrafficState, topupState)) Future.successful(Seq.empty) + else { + val shortfall = + TopupMemberTrafficTrigger.shortfall(currentTrafficState.extraTrafficRemainder) + val shortfallAwaitingGrant = TopupMemberTrafficTrigger.shortfallAwaitingGrant( + currentTrafficState, + lastShortfallPurchases.get(target.synchronizerId), + ) + if (shortfallAwaitingGrant) + logger.info( + s"Not buying the ${BigDecimal(shortfall) / 1e6} MB of traffic this member is below zero on ${target.alias} " + + "with its top-up again: the participant has received no traffic update since that shortfall was bought, so that purchase is not granted yet" ) - .map(Some(_)) - } yield cost.map( - TopupMemberTrafficTrigger.Task( - target, - topupState, - currentTrafficState, - registration, - ) -> _ - ) - } yield task + // The cost is derived from this target's throughput and interval, plus the + // extra traffic bought with the top-up. + Future.traverse( + TopupMemberTrafficTrigger.extraTrafficOptions(shortfall, shortfallAwaitingGrant) + )(extraTraffic => + TopupUtil + .minWalletBalanceForTopup( + scanConnection, + target.topupConfig, + clock, + registration, + extraTraffic, + ) + .map( + TopupMemberTrafficTrigger.Task( + target, + topupState, + currentTrafficState, + registration, + extraTraffic, + shortfallAwaitingGrant, + ) -> _ + ) + ) + } + } yield tasks + } yield purchases override protected def completeTask( task: TopupMemberTrafficTrigger.Task )(implicit tc: TraceContext): Future[TaskOutcome] = { val coBuyMemberTraffic = new CO_BuyMemberTraffic( - task.target.topupParameters.topupAmount, + task.trafficToBuy, task.topupState.payload.memberId, task.topupState.payload.synchronizerId, task.topupState.payload.migrationId, @@ -208,6 +274,9 @@ class TopupMemberTrafficTrigger( ) .map { case outcome: COO_BuyMemberTraffic => + // Only once bought, so a purchase that fails is retried with the shortfall. + if (task.extraTraffic > 0) + lastShortfallPurchases.update(task.target.synchronizerId, task.trafficState) TaskSuccess(s"Successfully bought extra traffic: $outcome") case error: COO_Error => throw Status.ABORTED @@ -354,23 +423,47 @@ object TopupMemberTrafficTrigger { ) } - /** Splits candidates into the ones the balance covers and the ones it does not, in target order. - * Every buy draws on the same wallet, so a target is only funded once the ones before it are. - * A `None` budget does not bound the purchases. + /** Splits candidates into the purchases the balance covers and the candidates it covers none + * of, both in target order. Each candidate lists its purchases preferred first, and every buy + * draws on the same wallet, so the balance is allocated in two passes, each in target order. + * The first gives each candidate its first purchase that the balance covers and that + * `isUnblocking` accepts, one that takes its member's extra-traffic remainder above zero. The + * second gives each candidate still unfunded its first purchase the rest covers, such as a + * top-up that pays a shortfall down without covering it, or, while an earlier shortfall purchase + * is not granted yet, a top-up bought alone that is smaller than that shortfall, as it is ranked + * by the traffic state before that grant. A purchase that leaves its member below zero thus + * cannot take the balance from one that keeps a later target above it. A candidate the + * balance covers none of is reported by its last, cheapest, purchase. A `None` budget funds each + * candidate's first purchase. */ private[automation] def fundedTasks[A]( - candidates: Seq[(A, BigDecimal)], + candidates: Seq[Seq[(A, BigDecimal)]], budget: Option[BigDecimal], - ): (Seq[A], Seq[A]) = budget match { - case None => (candidates.map(_._1), Seq.empty) + )(isUnblocking: A => Boolean): (Seq[A], Seq[A]) = budget match { + case None => (candidates.flatMap(_.headOption.map(_._1)), Seq.empty) case Some(balance) => - val (funded, unfunded, _) = - candidates.foldLeft((Seq.empty[A], Seq.empty[A], balance)) { - case ((funded, unfunded, remaining), (task, cost)) => - if (cost <= remaining) (funded :+ task, unfunded, remaining - cost) - else (funded, unfunded :+ task, remaining) + // One pass in target order: each candidate not funded yet gets its first purchase that the + // balance left covers and that is `eligible`. + def fund( + chosen: Seq[Option[A]], + balanceLeft: BigDecimal, + eligible: A => Boolean, + ): (Seq[Option[A]], BigDecimal) = + candidates.zip(chosen).foldLeft((Seq.empty[Option[A]], balanceLeft)) { + case ((done, remaining), (_, Some(task))) => (done :+ Some(task), remaining) + case ((done, remaining), (purchases, None)) => + purchases.find { case (task, cost) => cost <= remaining && eligible(task) } match { + case Some((task, cost)) => (done :+ Some(task), remaining - cost) + case None => (done :+ None, remaining) + } } - (funded, unfunded) + val (unblocking, balanceLeft) = + fund(candidates.map(_ => Option.empty[A]), balance, isUnblocking) + val (funded, _) = fund(unblocking, balanceLeft, _ => true) + val unfunded = candidates.zip(funded).collect { case (purchases, None) => + purchases.lastOption.map(_._1) + } + (funded.flatten, unfunded.flatten) } /** Resolves the configured top-up targets against the synchronizers the participant is @@ -422,6 +515,43 @@ object TopupMemberTrafficTrigger { // Two aliases can resolve to one synchronizer; topupTargets is global-first, so global wins. .distinctBy(_.synchronizerId) + /** Traffic below zero, as once a dedicated synchronizer's operator removes an outage traffic + * allowance the member used. The member cannot submit until its purchases cover it. + */ + def shortfall(extraTrafficRemainder: Long): Long = math.max(0L, -extraTrafficRemainder) + + /** Whether buying `trafficToBuy` takes a member's extra-traffic remainder of + * `extraTrafficRemainder` strictly above zero. It ignores the base traffic and what a + * submission costs, so it only approximates whether the member can submit again, which is + * enough to rank purchases. + */ + def unblocks(trafficToBuy: Long, extraTrafficRemainder: Long): Boolean = + trafficToBuy > shortfall(extraTrafficRemainder) + + /** Whether the shortfall `trafficState` shows is already bought, by a purchase not granted yet: + * the target's last shortfall purchase, `lastShortfallPurchase`, bought the shortfall of a + * traffic state with the same serial. Each grant moves the serial, as the participant only + * applies a traffic update with a higher one, so no grant has reached the participant since. + * Any grant ends this, also one of some other purchase while ours is still pending. + */ + private[automation] def shortfallAwaitingGrant( + trafficState: TrafficState, + lastShortfallPurchase: Option[TrafficState], + ): Boolean = + shortfall(trafficState.extraTrafficRemainder) > 0 && + lastShortfallPurchase.exists(_.serial == trafficState.serial) + + /** The extra traffic each purchase due buys with the top-up, preferred first: the member's + * shortfall, and 0 for the top-up alone, which still pays a shortfall down one interval at a + * time. Only the top-up alone while the shortfall is awaiting the grant of an earlier purchase, + * so that it is not bought twice. + */ + private[automation] def extraTrafficOptions( + shortfall: Long, + shortfallAwaitingGrant: Boolean, + ): Seq[Long] = + if (shortfall > 0 && !shortfallAwaitingGrant) Seq(shortfall, 0L) else Seq(0L) + final case class Task( target: Target, topupState: Contract[ValidatorTopUpState.ContractId, ValidatorTopUpState], @@ -429,12 +559,31 @@ object TopupMemberTrafficTrigger { registration: Option[ ContractWithState[RegisteredSynchronizer.ContractId, RegisteredSynchronizer] ], + // Bought together with the configured top-up: the member's shortfall, so that one purchase + // covers it, or 0 for the top-up alone. That is every ordinary top-up, the fallback where + // the wallet does not cover both, and a top-up while the shortfall is awaiting a grant. + extraTraffic: Long, + // Whether the member's shortfall is already bought, by a purchase not granted yet. + shortfallAwaitingGrant: Boolean, ) extends PrettyPrinting { + + def shortfall: Long = TopupMemberTrafficTrigger.shortfall(trafficState.extraTrafficRemainder) + + def trafficToBuy: Long = target.topupParameters.topupAmount + extraTraffic + + /** Whether this purchase, once granted, takes the member's extra-traffic remainder strictly + * above zero. + */ + def unblocks: Boolean = + TopupMemberTrafficTrigger.unblocks(trafficToBuy, trafficState.extraTrafficRemainder) + override def pretty: Pretty[Task] = prettyOfClass[Task]( param("target", _.target), param("topupState", _.topupState), param("trafficState", _.trafficState), + param("extraTraffic", _.extraTraffic), + param("shortfallAwaitingGrant", _.shortfallAwaitingGrant), ) } } diff --git a/apps/validator/src/main/scala/org/lfdecentralizedtrust/splice/validator/config/ValidatorAppConfig.scala b/apps/validator/src/main/scala/org/lfdecentralizedtrust/splice/validator/config/ValidatorAppConfig.scala index 742813095d5..1e0566e7e34 100644 --- a/apps/validator/src/main/scala/org/lfdecentralizedtrust/splice/validator/config/ValidatorAppConfig.scala +++ b/apps/validator/src/main/scala/org/lfdecentralizedtrust/splice/validator/config/ValidatorAppConfig.scala @@ -53,6 +53,10 @@ final case class BuyExtraTrafficConfig( /** target throughput in bytes per second * * The top-up trigger uses this to determine how much extra traffic to purchase each time. + * Where the extra traffic balance is below zero, as once a dedicated synchronizer's operator + * removes an outage traffic allowance the validator used (traffic past what was bought, + * allowed while the Global Synchronizer is unavailable), a top-up also buys that shortfall if + * the wallet covers both. * A value of zero implies that no extra traffic will be purchased automatically and * the throughput will be limited to the base rate. */ @@ -62,8 +66,9 @@ final case class BuyExtraTrafficConfig( /** minimum interval between extra traffic purchases in seconds * * This allows validator operators to control the frequency at which the top-up trigger - * will charge them domain fees making spends more predictable. This should be greater than the - * polling interval of the top-up trigger. + * will charge them domain fees making spends more predictable, though a top-up that also buys + * a shortfall (see `targetThroughput`) costs more than the others. This should be greater + * than the polling interval of the top-up trigger. * * Note that the actual interval between top-ups might be larger due to the DSO requiring * a minimal top-up amount larger than `targetThroughput * minTopupInterval`. diff --git a/apps/validator/src/test/scala/org/lfdecentralizedtrust/splice/validator/automation/TopupMemberTrafficTriggerTest.scala b/apps/validator/src/test/scala/org/lfdecentralizedtrust/splice/validator/automation/TopupMemberTrafficTriggerTest.scala index 7e06f67f5cc..9e7ab33c763 100644 --- a/apps/validator/src/test/scala/org/lfdecentralizedtrust/splice/validator/automation/TopupMemberTrafficTriggerTest.scala +++ b/apps/validator/src/test/scala/org/lfdecentralizedtrust/splice/validator/automation/TopupMemberTrafficTriggerTest.scala @@ -4,7 +4,13 @@ package org.lfdecentralizedtrust.splice.validator.automation import com.digitalasset.canton.config.NonNegativeFiniteDuration -import com.digitalasset.canton.config.RequireTypes.NonNegativeNumeric +import com.digitalasset.canton.config.RequireTypes.{ + NonNegativeLong, + NonNegativeNumeric, + PositiveInt, +} +import com.digitalasset.canton.data.CantonTimestamp +import com.digitalasset.canton.sequencing.protocol.TrafficState import com.digitalasset.canton.topology.SynchronizerId import com.digitalasset.canton.{BaseTest, SynchronizerAlias} import org.lfdecentralizedtrust.splice.validator.config.BuyExtraTrafficConfig @@ -109,21 +115,194 @@ class TopupMemberTrafficTriggerTest extends AnyWordSpec with BaseTest { "TopupMemberTrafficTrigger.fundedTasks" should { - // Costs only; what they are attached to does not affect the allocation. - val candidates = Seq(globalAlias -> BigDecimal(100), dedicatedAlias -> BigDecimal(100)) + // A purchase, and whether it takes its member's extra-traffic remainder above zero. What else + // it is attached to does not affect the allocation. + final case class Purchase(name: String, unblocks: Boolean = true) + def fundedTasks(candidates: Seq[Seq[(Purchase, BigDecimal)]], budget: Option[BigDecimal]) = + TopupMemberTrafficTrigger.fundedTasks(candidates, budget)(_.unblocks) + + val global = Purchase("global") + val dedicated = Purchase("dedicated") + val candidates = Seq(Seq(global -> BigDecimal(100)), Seq(dedicated -> BigDecimal(100))) "fund only what the balance covers, in order" in { // Enough for either one alone, not for both, so the second one misses out. - val (funded, unfunded) = - TopupMemberTrafficTrigger.fundedTasks(candidates, Some(BigDecimal(150))) - funded shouldBe Seq(globalAlias) - unfunded shouldBe Seq(dedicatedAlias) + fundedTasks(candidates, Some(BigDecimal(150))) shouldBe (Seq(global), Seq(dedicated)) } "fund every target when the balance does not bound the purchases" in { - val (funded, unfunded) = TopupMemberTrafficTrigger.fundedTasks(candidates, None) - funded shouldBe Seq(globalAlias, dedicatedAlias) - unfunded shouldBe empty + fundedTasks(candidates, None) shouldBe (Seq(global, dedicated), Seq.empty) + } + + // A member below zero: its shortfall with the top-up, or the top-up alone, which leaves it + // at or below zero where the shortfall is at least the top-up. + val topupAndShortfall = Purchase("top-up and shortfall") + val topupBelowZero = Purchase("top-up", unblocks = false) + val withShortfall = + Seq(topupAndShortfall -> BigDecimal(300), topupBelowZero -> BigDecimal(100)) + + "buy the shortfall with the top-up where the balance covers both" in { + fundedTasks(Seq(withShortfall), Some(BigDecimal(300))) shouldBe + (Seq(topupAndShortfall), Seq.empty) + } + + "fall back to the top-up alone where the balance does not cover the shortfall too" in { + fundedTasks(Seq(withShortfall), Some(BigDecimal(150))) shouldBe + (Seq(topupBelowZero), Seq.empty) + } + + "report the top-up alone where the balance does not cover even that" in { + fundedTasks(Seq(withShortfall), Some(BigDecimal(50))) shouldBe + (Seq.empty, Seq(topupBelowZero)) + } + + "buy the shortfall with the top-up where the balance does not bound the purchases" in { + fundedTasks(Seq(withShortfall), None) shouldBe (Seq(topupAndShortfall), Seq.empty) + } + + // A member 500 below zero with a top-up of 100, ahead of a healthy target whose top-up costs + // 100 too. + val blocked = Seq( + Purchase("blocked: top-up and shortfall") -> BigDecimal(600), + Purchase("blocked: top-up", unblocks = false) -> BigDecimal(100), + ) + val healthy = Purchase("healthy") + val healthyTarget = Seq(healthy -> BigDecimal(100)) + + "fund a later target before a top-up that leaves an earlier member below zero" in { + // Enough for one top-up: the healthy target's keeps it submitting, the other would not get + // its member submitting again. + fundedTasks(Seq(blocked, healthyTarget), Some(BigDecimal(150))) shouldBe + (Seq(healthy), Seq(blocked.last._1)) + } + + "fund a top-up that leaves its member below zero with what the other targets leave" in { + fundedTasks(Seq(blocked, healthyTarget), Some(BigDecimal(250))) shouldBe + (Seq(blocked.last._1, healthy), Seq.empty) + } + + "fund a top-up that covers the shortfall alone ahead of later targets" in { + // A shortfall smaller than the top-up, so the top-up alone gets the member submitting again. + val coveredByTopup = Purchase("top-up covering the shortfall") + val smallShortfall = Seq( + Purchase("top-up and small shortfall") -> BigDecimal(120), + coveredByTopup -> BigDecimal(100), + ) + // Enough for one top-up, which goes to the first target. + fundedTasks(Seq(smallShortfall, healthyTarget), Some(BigDecimal(110))) shouldBe + (Seq(coveredByTopup), Seq(healthy)) + } + + "fund each candidate's first purchase where the balance does not bound the purchases" in { + fundedTasks(Seq(blocked, healthyTarget), None) shouldBe + (Seq(blocked.head._1, healthy), Seq.empty) + } + } + + "TopupMemberTrafficTrigger.shortfall" should { + + "be what a balance below zero needs to reach zero" in { + TopupMemberTrafficTrigger.shortfall(-300L) shouldBe 300L + } + + "be zero for a balance at or above zero" in { + TopupMemberTrafficTrigger.shortfall(0L) shouldBe 0L + TopupMemberTrafficTrigger.shortfall(500L) shouldBe 0L + } + } + + "TopupMemberTrafficTrigger.unblocks" should { + + "hold for a purchase that takes the balance above zero" in { + TopupMemberTrafficTrigger.unblocks( + trafficToBuy = 2_300L, + extraTrafficRemainder = -300L, + ) shouldBe true + TopupMemberTrafficTrigger.unblocks( + trafficToBuy = 301L, + extraTrafficRemainder = -300L, + ) shouldBe true + } + + "not hold for a purchase that leaves the balance at or below zero" in { + TopupMemberTrafficTrigger.unblocks( + trafficToBuy = 300L, + extraTrafficRemainder = -300L, + ) shouldBe false + TopupMemberTrafficTrigger.unblocks( + trafficToBuy = 2_000L, + extraTrafficRemainder = -2_001L, + ) shouldBe false + } + + "hold for any purchase at a balance at or above zero" in { + TopupMemberTrafficTrigger.unblocks( + trafficToBuy = 2_000L, + extraTrafficRemainder = 0L, + ) shouldBe true + TopupMemberTrafficTrigger.unblocks( + trafficToBuy = 2_000L, + extraTrafficRemainder = 500L, + ) shouldBe true + } + } + + "TopupMemberTrafficTrigger.shortfallAwaitingGrant" should { + + def trafficState(purchased: Long, consumed: Long, serial: Int) = + TrafficState + .empty(CantonTimestamp.Epoch) + .copy( + extraTrafficPurchased = NonNegativeLong.tryCreate(purchased), + extraTrafficConsumed = NonNegativeLong.tryCreate(consumed), + serial = Some(PositiveInt.tryCreate(serial)), + ) + + // 500 below zero, the state whose shortfall was bought. + val boughtFor = trafficState(purchased = 1_000L, consumed = 1_500L, serial = 3) + + "hold while the participant has received no traffic update since the shortfall was bought" in { + TopupMemberTrafficTrigger.shortfallAwaitingGrant(boughtFor, Some(boughtFor)) shouldBe true + } + + "not hold once a grant has moved the serial, even with the member still below zero" in { + // A grant of something else, +200, that leaves the member 300 below zero: the hold ends, + // and those 300 are bought again. + val updated = trafficState(purchased = 1_200L, consumed = 1_500L, serial = 4) + TopupMemberTrafficTrigger.shortfallAwaitingGrant(updated, Some(boughtFor)) shouldBe false + } + + "not hold with no shortfall purchase on record, as after a restart" in { + TopupMemberTrafficTrigger.shortfallAwaitingGrant(boughtFor, None) shouldBe false + } + + "not hold for a member that is not below zero" in { + val aboveZero = trafficState(purchased = 2_000L, consumed = 1_500L, serial = 3) + TopupMemberTrafficTrigger.shortfallAwaitingGrant(aboveZero, Some(boughtFor)) shouldBe false + } + } + + "TopupMemberTrafficTrigger.extraTrafficOptions" should { + + "offer the shortfall with the top-up first, then the top-up alone" in { + TopupMemberTrafficTrigger.extraTrafficOptions( + shortfall = 500L, + shortfallAwaitingGrant = false, + ) shouldBe Seq(500L, 0L) + } + + "offer the top-up alone while the grant of an earlier shortfall purchase is pending" in { + TopupMemberTrafficTrigger.extraTrafficOptions( + shortfall = 500L, + shortfallAwaitingGrant = true, + ) shouldBe Seq(0L) + } + + "offer only the top-up alone for a member that is not below zero" in { + TopupMemberTrafficTrigger.extraTrafficOptions( + shortfall = 0L, + shortfallAwaitingGrant = false, + ) shouldBe Seq(0L) } } } diff --git a/apps/wallet/src/main/scala/org/lfdecentralizedtrust/splice/wallet/util/TopupUtil.scala b/apps/wallet/src/main/scala/org/lfdecentralizedtrust/splice/wallet/util/TopupUtil.scala index a8c2b7c119c..25e2cc6008f 100644 --- a/apps/wallet/src/main/scala/org/lfdecentralizedtrust/splice/wallet/util/TopupUtil.scala +++ b/apps/wallet/src/main/scala/org/lfdecentralizedtrust/splice/wallet/util/TopupUtil.scala @@ -30,6 +30,7 @@ object TopupUtil { .map(_.payload.governanceParameters) .fold(BigDecimal(1))(p => BigDecimal(p.discountFactor)) + /** The cost of one top-up, plus `extraTraffic` bought with it. */ def minWalletBalanceForTopup( scanConnection: ScanConnection, validatorTopupConfig: ValidatorTopupConfig, @@ -37,6 +38,7 @@ object TopupUtil { registration: Option[ ContractWithState[RegisteredSynchronizer.ContractId, RegisteredSynchronizer] ], + extraTraffic: Long = 0L, )(implicit tc: TraceContext, ec: ExecutionContext, mat: Materializer): Future[BigDecimal] = for { amuletRules <- scanConnection.getAmuletRulesWithState() synchronizerFeesConfig = AmuletConfigSchedule(amuletRules) @@ -50,16 +52,32 @@ object TopupUtil { validatorTopupConfig.topupTriggerPollingInterval, ) latestRound <- scanConnection.getLatestOpenMiningRound() - amuletPrice = latestRound.payload.amuletPrice - extraTrafficPrice = BigDecimal(synchronizerFeesConfig.extraTrafficPrice) - } yield SpliceUtil - .synchronizerFees( - topupParameters.topupAmount, - extraTrafficPrice, - amuletPrice, - discountFactor(registration), - ) - ._2 + } yield topupCost( + topupParameters, + extraTraffic, + BigDecimal(synchronizerFeesConfig.extraTrafficPrice), + latestRound.payload.amuletPrice, + registration, + ) + + /** What one top-up, plus `extraTraffic` bought with it, costs in amulet. */ + def topupCost( + topupParameters: ExtraTrafficTopupParameters, + extraTraffic: Long, + extraTrafficPrice: BigDecimal, + amuletPrice: BigDecimal, + registration: Option[ + ContractWithState[RegisteredSynchronizer.ContractId, RegisteredSynchronizer] + ], + ): BigDecimal = + SpliceUtil + .synchronizerFees( + topupParameters.topupAmount + extraTraffic, + extraTrafficPrice, + amuletPrice, + discountFactor(registration), + ) + ._2 private def currentWalletBalance(scanConnection: ScanConnection, store: UserWalletStore)(implicit tc: TraceContext, diff --git a/apps/wallet/src/test/scala/org/lfdecentralizedtrust/splice/wallet/util/TopupUtilTest.scala b/apps/wallet/src/test/scala/org/lfdecentralizedtrust/splice/wallet/util/TopupUtilTest.scala new file mode 100644 index 00000000000..89906b9b08a --- /dev/null +++ b/apps/wallet/src/test/scala/org/lfdecentralizedtrust/splice/wallet/util/TopupUtilTest.scala @@ -0,0 +1,40 @@ +// Copyright (c) 2024 Digital Asset (Switzerland) GmbH and/or its affiliates. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +package org.lfdecentralizedtrust.splice.wallet.util + +import com.digitalasset.canton.BaseTestWordSpec +import com.digitalasset.canton.config.NonNegativeFiniteDuration + +class TopupUtilTest extends BaseTestWordSpec { + + // 2,000 bytes per top-up at $60 per MB is $0.12, or 24 amulets at $0.005 each. + private val topupParameters = + ExtraTrafficTopupParameters(2_000L, NonNegativeFiniteDuration.ofMinutes(1)) + private val extraTrafficPrice = BigDecimal(60) + private val amuletPrice = BigDecimal("0.005") + + "TopupUtil.topupCost" should { + + "price one top-up" in { + TopupUtil.topupCost( + topupParameters, + extraTraffic = 0L, + extraTrafficPrice, + amuletPrice, + registration = None, + ) shouldBe BigDecimal(24) + } + + "price a shortfall bought with the top-up as part of it" in { + // 5,000 bytes in all, so the funds check covers everything the purchase buys. + TopupUtil.topupCost( + topupParameters, + extraTraffic = 3_000L, + extraTrafficPrice, + amuletPrice, + registration = None, + ) shouldBe BigDecimal(60) + } + } +} diff --git a/docs/src/common/traffic_topups.rst b/docs/src/common/traffic_topups.rst index fa17bccbd31..d2e5b1daabc 100644 --- a/docs/src/common/traffic_topups.rst +++ b/docs/src/common/traffic_topups.rst @@ -4,7 +4,7 @@ SPDX-License-Identifier: Apache-2.0 On each successful top-up, the validator app purchases a `top-up amount` of roughly ``targetThroughput * minTopupInterval`` bytes of traffic -(specific amount can vary due to rounding-up). +(specific amount can vary due to rounding-up), plus any shortfall described below. The ``minTopupInterval`` allows validator operators to control the upper-bound frequency at which automated top-ups happen. If the top-up amount is below the synchronizer-wide ``minTopupAmount`` (see :ref:`traffic_parameters`), ``minTopupInterval`` is automatically stretched so that at least ``minTopupAmount`` bytes of traffic are @@ -17,6 +17,13 @@ The next top-up gets triggered when all of the following conditions are met: - At least ``minTopupInterval`` has elapsed since the last top-up. - The validator has sufficient CC in its wallet to buy the top-up amount worth on traffic (except on DevNet, where the validator app will automatically tap enough coin to purchase traffic). + A shortfall is only bought along with it if the wallet covers both. + +If the validator's extra traffic balance on a synchronizer is below zero, a top-up also buys that shortfall, +so that a single purchase lets the validator submit there again. +This can happen on a dedicated synchronizer whose operator, while the Global Synchronizer is unavailable, +lets validators use more traffic than they bought (an outage traffic allowance), +and then removes that allowance after a validator used it. Validators receive a small amount of free traffic from the Super Validators, which suffices for submitting the diff --git a/test-full-class-names-non-integration.log b/test-full-class-names-non-integration.log index 392ca705358..d199d50ee60 100644 --- a/test-full-class-names-non-integration.log +++ b/test-full-class-names-non-integration.log @@ -75,4 +75,5 @@ org.lfdecentralizedtrust.splice.validator.automation.TopupMemberTrafficTriggerTe org.lfdecentralizedtrust.splice.validator.store.DbValidatorConfigProviderTest org.lfdecentralizedtrust.splice.validator.store.DbValidatorStoreTest org.lfdecentralizedtrust.splice.wallet.util.ExtraTrafficTopupParametersTest +org.lfdecentralizedtrust.splice.wallet.util.TopupUtilTest org.lfdecentralizedtrust.splice.wart.ParTraverseTest