feat: buy a member's shortfall with its top-up after an outage allowance is removed

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

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`.

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

apps/wallet/src/main/scala/org/lfdecentralizedtrust/splice/wallet/util/TopupUtil.scala

@@ -30,13 +30,15 @@ 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,
       clock: Clock,
       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,