diff --git a/apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/EnvironmentDefinition.scala b/apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/EnvironmentDefinition.scala
index 34a34251a58..5f98ad7c30b 100644
--- a/apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/EnvironmentDefinition.scala
+++ b/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())
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 5c0f50fb33f..82f6f7a6fc3 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
@@ -4,12 +4,19 @@
 package org.lfdecentralizedtrust.splice.integration.tests
 
 import com.digitalasset.canton.SynchronizerAlias
+import com.digitalasset.canton.config.CantonRequireTypes.InstanceName
 import com.digitalasset.canton.config.NonNegativeFiniteDuration
-import com.digitalasset.canton.config.RequireTypes.{NonNegativeLong, NonNegativeNumeric}
-import com.digitalasset.canton.topology.SynchronizerId
+import com.digitalasset.canton.config.RequireTypes.{
+  NonNegativeLong,
+  NonNegativeNumeric,
+  PositiveLong,
+}
+import com.digitalasset.canton.logging.SuppressionRule
+import com.digitalasset.canton.topology.{Member, SynchronizerId}
 import monocle.macros.syntax.lens.*
 import org.lfdecentralizedtrust.splice.codegen.java.splice.decentralizedsynchronizer.GovernanceParameters
 import org.lfdecentralizedtrust.splice.codegen.java.splice.wallet.topupstate.ValidatorTopUpState
+import org.lfdecentralizedtrust.splice.console.ParticipantClientReference
 import org.lfdecentralizedtrust.splice.environment.TopologyAdminConnection.TopologySnapshot
 import org.lfdecentralizedtrust.splice.config.ConfigTransforms
 import org.lfdecentralizedtrust.splice.http.v0.definitions as d0
@@ -26,12 +33,15 @@ import org.lfdecentralizedtrust.splice.util.{
 }
 import org.lfdecentralizedtrust.splice.validator.automation.TopupMemberTrafficTrigger
 import org.lfdecentralizedtrust.splice.wallet.store.TxLogEntry
+import org.slf4j.event.Level
 
 import scala.concurrent.duration.*
 import scala.jdk.CollectionConverters.*
+import scala.util.Try
 
 /** Buys traffic for a registered synchronizer via `AmuletRules_BuyMemberTraffic` with the
-  * registration disclosed, and checks the operator grants it on that synchronizer's sequencer.
+  * registration disclosed, and checks the operator grants it on that synchronizer's sequencer,
+  * also through a global-synchronizer outage on the operator's outage traffic allowance.
   */
 @org.lfdecentralizedtrust.splice.util.scalatesttags.SpliceDsoGovernance_0_1_30
 class SyncOperatorTrafficIntegrationTest
@@ -46,6 +56,17 @@ class SyncOperatorTrafficIntegrationTest
   private val walletRequestPurchase = 3_000_000L
   private val discount = BigDecimal("0.5")
   private val splitwellAlias = SynchronizerAlias.tryCreate("splitwell")
+  private val globalAlias = SynchronizerAlias.tryCreate("global")
+
+  // The outage test's amounts: small enough that a few dozen pings take bob past what was bought
+  // for him and on through the allowance.
+  private val outageTrafficAllowance = 50_000L
+  private val bobPurchase = 50_000L
+  // Bought once the global synchronizer is back but before the allowance is removed.
+  private val bobPurchaseWithAllowanceSet = 20_000L
+  // The same operator app with the allowance set, which the outage test runs in place of
+  // syncOperator. Both use one store, so they never run at the same time.
+  private val syncOperatorWithAllowanceName = "syncOperatorWithAllowance"
 
   override def environmentDefinition: SpliceEnvironmentDefinition =
     EnvironmentDefinition
@@ -53,14 +74,29 @@ class SyncOperatorTrafficIntegrationTest
         Seq("simple-topology-1sv.conf", "sync-operator-topology.conf"),
         this.getClass.getSimpleName,
       )
-      .withOnlyAliceValidatorConnectingToSplitwell
+      // Only alice and bob connect to splitwell: alice for the purchases, bob for the outage test,
+      // which needs a member whose balance starts small. The operator's participant joins during
+      // the outage test, as a member with no purchase on record.
+      .withValidatorsDisconnectedFromSplitwell("splitwellValidator")
       .withStandardSetup
-      // withStandardSetup turns top-ups on for the decentralized synchronizer. Move alice's
-      // target onto splitwell instead, so the operator's synchronizer is the only one she tops
-      // up automatically.
+      // Down from 200 KB, so bob's small purchases are allowed.
+      .addConfigTransform((_, conf) =>
+        ConfigTransforms.updateAllSvAppFoundDsoConfigs_(
+          _.focus(_.initialSynchronizerFeesConfig.minTopupAmount)
+            .replace(NonNegativeLong.tryCreate(10_000L))
+        )(conf)
+      )
+      // withStandardSetup turns top-ups on for the decentralized synchronizer. Move alice's and
+      // bob's targets onto splitwell instead, so the operator's synchronizer is the only one they
+      // top up automatically.
       .addConfigTransform((_, config) =>
         ConfigTransforms.updateAllValidatorConfigs { case (name, validatorConfig) =>
-          if (name == "aliceValidator")
+          if (name == "bobValidator")
+            validatorConfig
+              // Paused throughout, so bob's traffic on splitwell is only what the test buys him.
+              .focus(_.automation)
+              .modify(_.withPausedTrigger[TopupMemberTrafficTrigger])
+          else if (name == "aliceValidator")
             validatorConfig
               .focus(_.domains.global.buyExtraTraffic.targetThroughput)
               .replace(NonNegativeNumeric.tryCreate(BigDecimal(0)))
@@ -74,15 +110,37 @@ class SyncOperatorTrafficIntegrationTest
                     .replace(NonNegativeFiniteDuration.ofMinutes(1))
                 else extra
               })
-              // Resumed only at the end of the test: a trigger buying on splitwell for the same
-              // member would break the exact-equality assertions on the two manual purchases.
-              // Alice's validator only: pausing a trigger an app does not register warns, and the
-              // SV validator never registers this one.
+              // Resumed only at the end of the first test: a trigger buying on splitwell for the
+              // same member would break the exact-equality assertions on the two manual purchases.
+              // Alice's and bob's validators only: pausing a trigger an app does not register warns,
+              // and the SV validator never registers this one.
               .focus(_.automation)
               .modify(_.withPausedTrigger[TopupMemberTrafficTrigger])
           else validatorConfig
         }(config)
       )
+      // Last, so it is a copy of syncOperator as every transform above left it.
+      .addConfigTransform { (_, conf) =>
+        val allowance = Some(PositiveLong.tryCreate(outageTrafficAllowance))
+        val syncOperator = conf.syncOperatorApps(InstanceName.tryCreate("syncOperator"))
+        conf.copy(syncOperatorApps =
+          conf.syncOperatorApps.updated(
+            InstanceName.tryCreate(syncOperatorWithAllowanceName),
+            syncOperator.copy(outageTrafficAllowance = allowance),
+          )
+        )
+      }
+      // Every app but syncOperatorWithAllowance starts as it would without manual start, and the
+      // multi-synchronizer flag that withManualStart's reset drops is set again once they have.
+      .withManualStart
+      .withAdditionalSetup(implicit env =>
+        startAllSync(
+          env.amuletNodes.local
+            .distinctBy(_.name)
+            .filterNot(_.name == syncOperatorWithAllowanceName)*
+        )
+      )
+      .withMultiSyncFeatureFlag()
 
   "sync operator" should {
 
@@ -271,8 +329,238 @@ class SyncOperatorTrafficIntegrationTest
         }
       }
     }
+
+    "keep members transacting through a global-synchronizer outage on the operator's allowance" in {
+      implicit env =>
+        val synchronizerId = syncOperatorBackend.appState.store.key.synchronizerId
+        val dsoParty = sv1Backend.getDsoInfo().dsoParty
+        // Registered by the test above.
+        val registration = eventually() {
+          sv1ScanBackend.lookupSynchronizerRegistration(synchronizerId.toProtoPrimitive).value
+        }
+        val aliceParticipant = aliceValidatorBackend.participantClientWithAdminToken
+        val alice = aliceParticipant.id
+        val bobParticipant = bobValidatorBackend.participantClientWithAdminToken
+        val bob = bobParticipant.id
+        // The operator app sees purchases, which are made on the global synchronizer, through
+        // this participant.
+        val operatorParticipant = splitwellValidatorBackend.participantClientWithAdminToken
+        // Not connected to splitwell until it joins during the outage.
+        val operator = operatorParticipant.id
+        val syncOperatorWithAllowance = syncop(syncOperatorWithAllowanceName)
+        def purchased() =
+          runningSyncOperatorBackend.appState.store.listTotalPurchasedMemberTraffic().futureValue
+        def limitsAreTotalsPlus(allowance: Long, totals: Map[Member, Long]) =
+          forAll(totals) { case (member, total) =>
+            extraTrafficLimit(member) shouldBe total + allowance
+          }
+
+        clue("bob holds no traffic on splitwell yet") {
+          extraTrafficLimit(bob) shouldBe 0L
+        }
+
+        // Anyone can fund a member's traffic, so alice pays for bob's.
+        val aliceParty = onboardWalletUser(aliceWalletClient, aliceValidatorBackend)
+        aliceWalletClient.tap(walletUsdToAmulet(50.0))
+        actAndCheck(
+          "alice buys a little traffic for bob on splitwell",
+          buyTraffic(aliceParty, bob, synchronizerId, Some(registration), dsoParty, bobPurchase),
+        )(
+          "the purchase is granted",
+          _ => extraTrafficLimit(bob) shouldBe bobPurchase,
+        )
+        val totals = purchased()
+        totals.keySet shouldBe Set(alice, bob)
+
+        try {
+          // Suppresses only what is expected while the operator's participant is cut off from the
+          // global synchronizer, so the sync operator apps' own warnings and errors still show:
+          // the warnings of the operator's validator, whose triggers fail meanwhile, and the
+          // errors the test console (created in EnvironmentDefinition) logs for bob's refused
+          // pings.
+          loggerFactory.suppress(
+            (SuppressionRule.LoggerNameContains("validator=splitwellValidator") &&
+              SuppressionRule.Level(Level.WARN)) ||
+              (SuppressionRule.forLogger[EnvironmentDefinition] &&
+                SuppressionRule.Level(Level.ERROR))
+          ) {
+            // Cut off from the global synchronizer, the operator's participant sees no purchases,
+            // which is what an outage looks like to the operator.
+            operatorParticipant.synchronizers.disconnect(globalAlias)
+
+            actAndCheck(
+              "the operator sets the allowance and restarts its app", {
+                syncOperatorBackend.stop()
+                syncOperatorWithAllowance.startSync()
+              },
+            )(
+              "each member's limit rises to its purchased total plus the allowance",
+              _ => limitsAreTotalsPlus(outageTrafficAllowance, totals),
+            )
+
+            clue("a restart with the allowance still set moves no limit") {
+              // A grant moves the serial even where it sets the limit the member already has.
+              def serials() = totals.keys.map(member => member -> serial(member)).toMap
+              val serialsBefore = serials()
+              syncOperatorWithAllowance.stop()
+              syncOperatorWithAllowance.startSync()
+              // Long enough for several of the restarted app's polls.
+              val window = 10.seconds
+              val pollingInterval = syncOperatorWithAllowance.config.automation.pollingInterval
+              pollingInterval.asFiniteApproximation * 3 should be < window
+              always(durationOfSuccess = window, pollIntervalMs = 1000) {
+                limitsAreTotalsPlus(outageTrafficAllowance, totals)
+                serials() shouldBe serialsBefore
+              }
+            }
+
+            // The most one of bob's pings took his consumed traffic up by.
+            val pingCost = clue("bob keeps transacting past what was bought for him") {
+              val costs = pingWhile(bobParticipant, synchronizerId)(consumed(bob) <= bobPurchase)
+              consumed(bob) should be > bobPurchase
+              costs.maxOption.getOrElse(fail("bob never pinged"))
+            }
+
+            clue("and is refused once he has used the allowance") {
+              def headroom = bobPurchase + outageTrafficAllowance - consumed(bob)
+              // A ping can fail for other reasons too, so only a failure that leaves bob less than
+              // one ping's worth of traffic shows that he has used the allowance up. A few other
+              // failures are tolerated, rather than up to maxPings that can each take a timeout.
+              val refused = Iterator
+                .continually(Try(ping(bobParticipant, synchronizerId)))
+                .take(maxPings)
+                .filter(_.isFailure)
+                .take(maxRefusals)
+                .exists(_ => headroom < pingCost)
+              withClue(s"bob has $headroom left, and one ping took up to $pingCost:") {
+                refused shouldBe true
+              }
+            }
+
+            clue("alice runs on what was bought for her, without the allowance") {
+              val consumedBefore = consumed(alice)
+              ping(aliceParticipant, synchronizerId)
+              eventually() {
+                consumed(alice) should be > consumedBefore
+              }
+              consumed(alice) should be <= totals(alice)
+            }
+
+            actAndCheck(timeUntilSuccess = 1.minute)(
+              "the operator's participant joins splitwell, with no purchase on record",
+              operatorParticipant.synchronizers.connect_by_config(
+                aliceParticipant.synchronizers
+                  .config(splitwellAlias)
+                  .value
+                  // So that nothing reconnects it once it has left again.
+                  .copy(manualConnect = true),
+                // Its topology transactions only go through once it has traffic.
+                synchronize = None,
+              ),
+            )(
+              "its limit is the allowance alone",
+              _ => {
+                purchased().keySet should not contain operator
+                extraTrafficLimit(operator) shouldBe outageTrafficAllowance
+              },
+            )
+            // It only joined for that check.
+            operatorParticipant.synchronizers.disconnect(splitwellAlias)
+
+            operatorParticipant.synchronizers.reconnect(globalAlias)
+
+            actAndCheck(
+              "alice buys more traffic for bob once the global synchronizer is back",
+              buyTraffic(
+                aliceParty,
+                bob,
+                synchronizerId,
+                Some(registration),
+                dsoParty,
+                bobPurchaseWithAllowanceSet,
+              ),
+            )(
+              "bob's limit rises to his new purchased total plus the allowance",
+              _ => {
+                purchased()(bob) shouldBe bobPurchase + bobPurchaseWithAllowanceSet
+                extraTrafficLimit(bob) shouldBe
+                  bobPurchase + bobPurchaseWithAllowanceSet + outageTrafficAllowance
+              },
+            )
+
+            syncOperatorWithAllowance.stop()
+          }
+        } finally {
+          def connectedTo(alias: SynchronizerAlias) =
+            operatorParticipant.synchronizers.list_connected().exists(_.synchronizerAlias == alias)
+          if (connectedTo(splitwellAlias))
+            operatorParticipant.synchronizers.disconnect(splitwellAlias)
+          if (!connectedTo(globalAlias))
+            operatorParticipant.synchronizers.reconnect(globalAlias)
+        }
+
+        val totalsWithoutAllowance = actAndCheck(
+          "the operator restarts its app without the allowance", {
+            syncOperatorBackend.startSync()
+            purchased()
+          },
+        )(
+          "each member's limit goes back to its purchased total, which is zero for the participant " +
+            "that joined during the outage",
+          totals => {
+            limitsAreTotalsPlus(0L, totals)
+            extraTrafficLimit(operator) shouldBe 0L
+          },
+        )._1
+
+        clue("alice, who did not use the allowance, keeps transacting") {
+          ping(aliceParticipant, synchronizerId)
+        }
+
+        clue("bob used more than was bought for him, so he is refused until a purchase covers it") {
+          consumed(bob) should be > totalsWithoutAllowance(bob)
+          assertThrowsAndLogsCommandFailures(
+            ping(bobParticipant, synchronizerId),
+            _.errorMessage should include("Unable to ping"),
+          )
+        }
+
+    }
   }
 
+  // Bounds the ping loops, so a ping that costs less than expected fails the test instead of
+  // running forever.
+  private val maxPings = 300
+
+  // Bounds the failed pings the refusal check looks at.
+  private val maxRefusals = 5
+
+  private def ping(participant: ParticipantClientReference, synchronizerId: SynchronizerId) =
+    participant.health.ping(participant.id, synchronizerId = Some(synchronizerId))
+
+  /** Pings while `condition` holds, and returns how much each ping took the participant's
+    * consumed traffic up by.
+    */
+  private def pingWhile(participant: ParticipantClientReference, synchronizerId: SynchronizerId)(
+      condition: => Boolean
+  )(implicit env: SpliceTestConsoleEnvironment): Seq[Long] =
+    Iterator
+      .continually(())
+      .take(maxPings)
+      .takeWhile(_ => condition)
+      .map { _ =>
+        val before = consumed(participant.id)
+        ping(participant, synchronizerId)
+        consumed(participant.id) - before
+      }
+      .toSeq
+
+  private def consumed(member: Member)(implicit env: SpliceTestConsoleEnvironment): Long =
+    trafficState(member).fold(0L)(_.extraTrafficConsumed.value)
+
+  private def serial(member: Member)(implicit env: SpliceTestConsoleEnvironment) =
+    trafficState(member).flatMap(_.state.serial)
+
   // The sync-operator CI job bootstraps splitwell with traffic control and a zero base rate;
   // see start-canton.sh -t.
   private def synchronizerParameters(implicit env: SpliceTestConsoleEnvironment) =
diff --git a/apps/app/src/test/scala/org/lfdecentralizedtrust/splice/util/SyncOperatorTestUtil.scala b/apps/app/src/test/scala/org/lfdecentralizedtrust/splice/util/SyncOperatorTestUtil.scala
index f74537b0050..e26fee6fd91 100644
--- a/apps/app/src/test/scala/org/lfdecentralizedtrust/splice/util/SyncOperatorTestUtil.scala
+++ b/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
 
diff --git a/apps/common/src/main/scala/org/lfdecentralizedtrust/splice/automation/ReconcileSequencerLimitWithMemberTrafficTriggerBase.scala b/apps/common/src/main/scala/org/lfdecentralizedtrust/splice/automation/ReconcileSequencerLimitWithMemberTrafficTriggerBase.scala
index f35e5142205..f1af562c39b 100644
--- a/apps/common/src/main/scala/org/lfdecentralizedtrust/splice/automation/ReconcileSequencerLimitWithMemberTrafficTriggerBase.scala
+++ b/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.
   *
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
index 23dd4de7b7d..38968e4744c 100644
--- 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
@@ -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
+  }
 }
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 58893cc4b24..7ebd94956db 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
@@ -173,6 +173,7 @@ class SyncOperatorApp(
         config.lsu,
         config.lsuDumpPath,
         config.trafficBalanceReconciliationDelay,
+        config.outageTrafficAllowance,
         loggerFactory,
         packageVersionSupport,
       )
diff --git a/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/automation/ReconcileDedicatedSequencerTrafficTrigger.scala b/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/automation/ReconcileDedicatedSequencerTrafficTrigger.scala
index afb1e5b5f4d..83e8741745c 100644
--- a/apps/syncoperator/src/main/scala/org/lfdecentralizedtrust/splice/syncoperator/automation/ReconcileDedicatedSequencerTrafficTrigger.scala
+++ b/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),
+    )
+  }
 }
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 04c0b6a3df2..317fe8032ea 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
@@ -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,
     )
   )
 
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 33044bd6c12..1af10c32faf 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
@@ -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.
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 6e0756994bd..7f00ede977a 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
@@ -4,7 +4,7 @@
 package org.lfdecentralizedtrust.splice.syncoperator.store
 
 import com.digitalasset.canton.lifecycle.CloseContext
-import com.digitalasset.canton.logging.NamedLoggerFactory
+import com.digitalasset.canton.logging.{NamedLoggerFactory, TracedLogger}
 import com.digitalasset.canton.logging.pretty.{Pretty, PrettyPrinting}
 import com.digitalasset.canton.resource.DbStorage
 import com.digitalasset.canton.topology.{Member, ParticipantId, PartyId, SynchronizerId}
@@ -37,6 +37,13 @@ trait SyncOperatorStore extends AppStore {
   def getTotalPurchasedMemberTraffic(memberId: Member)(implicit
       tc: TraceContext
   ): Future[Long]
+
+  /** Total traffic purchased on this operator's synchronizer, per member with a purchase on
+    * record.
+    */
+  def listTotalPurchasedMemberTraffic()(implicit
+      tc: TraceContext
+  ): Future[Map[Member, Long]]
 }
 
 object SyncOperatorStore {
@@ -81,7 +88,7 @@ object SyncOperatorStore {
     )
   }
 
-  def contractFilter(key: Key): MultiDomainAcsStore.ContractFilter[
+  def contractFilter(key: Key, logger: TracedLogger): MultiDomainAcsStore.ContractFilter[
     SyncOperatorAcsStoreRowData,
     AcsInterfaceViewRowData.NoInterfacesIngested,
   ] = {
@@ -105,7 +112,17 @@ object SyncOperatorStore {
               .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(_)),
+              .fold(
+                error => {
+                  // The row encoder gets no trace context.
+                  logger.warn(
+                    s"Skipping the purchased traffic in ${contract.contractId.contractId} of " +
+                      s"member ${contract.payload.memberId}, which does not parse: $error"
+                  )(TraceContext.empty)
+                  None
+                },
+                Some(_),
+              ),
             // 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 8c8f126cb75..c5e134fb0ab 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
@@ -76,7 +76,7 @@ class DbSyncOperatorStore(
   override lazy val acsContractFilter: MultiDomainAcsStore.ContractFilter[
     SyncOperatorAcsStoreRowData,
     AcsInterfaceViewRowData.NoInterfacesIngested,
-  ] = SyncOperatorStore.contractFilter(key)
+  ] = SyncOperatorStore.contractFilter(key, logger)
 
   import multiDomainAcsStore.waitUntilAcsIngested
 
@@ -94,4 +94,16 @@ class DbSyncOperatorStore(
       key.synchronizerId,
     )
   }
+
+  override def listTotalPurchasedMemberTraffic()(implicit
+      tc: TraceContext
+  ): Future[Map[Member, Long]] = waitUntilAcsIngested {
+    sumPurchasedMemberTrafficPerMember(
+      storage,
+      SyncOperatorTables.acsTableName,
+      acsStoreId,
+      domainMigrationId,
+      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 3e2e038659b..0ddb14b8898 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
@@ -13,7 +13,7 @@ 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.store.{HardLimit, Limit, MultiDomainAcsStore, StoreTestBase}
 import org.lfdecentralizedtrust.splice.syncoperator.store.SyncOperatorStore
 import org.lfdecentralizedtrust.splice.syncoperator.store.db.DbSyncOperatorStore
 import org.lfdecentralizedtrust.splice.util.{Contract, ResourceTemplateDecoder, TemplateJsonDecoder}
@@ -41,10 +41,25 @@ abstract class SyncOperatorStoreTest extends StoreTestBase with HasExecutionCont
       synchronizerId: SynchronizerId = ourSynchronizer,
       operator: Option[PartyId] = Some(operatorParty),
       migrationId: Long = 0L,
+  ): Contract[MemberTraffic.ContractId, MemberTraffic] =
+    memberTrafficOfId(
+      member.toProtoPrimitive,
+      totalPurchased,
+      synchronizerId,
+      operator,
+      migrationId,
+    )
+
+  private def memberTrafficOfId(
+      memberId: String,
+      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,
+      memberId,
       synchronizerId.toProtoPrimitive,
       migrationId,
       totalPurchased,
@@ -63,8 +78,13 @@ abstract class SyncOperatorStoreTest extends StoreTestBase with HasExecutionCont
   private def ingest(
       store: SyncOperatorStore,
       contracts: Seq[Contract[MemberTraffic.ContractId, MemberTraffic]],
+  ): Future[Unit] = ingestInto(store.multiDomainAcsStore.testIngestionSink, contracts)
+
+  private def ingestInto(
+      sink: MultiDomainAcsStore.IngestionSink,
+      contracts: Seq[Contract[MemberTraffic.ContractId, MemberTraffic]],
   ): Future[Unit] =
-    store.multiDomainAcsStore.testIngestionSink.ingestAcs(
+    sink.ingestAcs(
       nextOffset(),
       // the purchase is made and recorded on the decentralized synchronizer, not on the dedicated
       // one it names
@@ -141,6 +161,46 @@ abstract class SyncOperatorStoreTest extends StoreTestBase with HasExecutionCont
         total <- store.getTotalPurchasedMemberTraffic(alice)
       } yield total shouldBe 0L
     }
+
+    "list the total of every member with a purchase for this synchronizer" in {
+      for {
+        store <- mkStore()
+        _ <- ingest(
+          store,
+          Seq(
+            memberTraffic(alice, 100L),
+            memberTraffic(alice, 250L),
+            memberTraffic(bob, 700L),
+            memberTraffic(bob, 900L, synchronizerId = foreignSynchronizer),
+            memberTraffic(bob, 300L, operator = Some(otherOperator)),
+          ),
+        )
+        totals <- store.listTotalPurchasedMemberTraffic()
+      } yield totals shouldBe Map(alice -> 350L, bob -> 700L)
+    }
+
+    "leave a purchase whose member id does not parse out of every total, with a warning" in {
+      for {
+        store <- mkStore()
+        _ <- loggerFactory.assertLogs(
+          // Past the test sink, which fails on any warning.
+          ingestInto(
+            store.multiDomainAcsStore.ingestionSink,
+            Seq(memberTraffic(alice, 100L), memberTrafficOfId("not-a-member", 250L)),
+          ),
+          _.warningMessage should include("of member not-a-member, which does not parse"),
+        )
+        totals <- store.listTotalPurchasedMemberTraffic()
+      } yield totals shouldBe Map(alice -> 100L)
+    }
+
+    "list no totals before any purchase" in {
+      for {
+        store <- mkStore()
+        _ <- ingest(store, Seq.empty)
+        totals <- store.listTotalPurchasedMemberTraffic()
+      } yield totals shouldBe empty
+    }
   }
 }
 
diff --git a/apps/syncoperator/src/test/scala/org/lfdecentralizedtrust/splice/syncoperator/automation/ReconcileDedicatedSequencerTrafficTriggerTest.scala b/apps/syncoperator/src/test/scala/org/lfdecentralizedtrust/splice/syncoperator/automation/ReconcileDedicatedSequencerTrafficTriggerTest.scala
new file mode 100644
index 00000000000..9a36c21c7bc
--- /dev/null
+++ b/apps/syncoperator/src/test/scala/org/lfdecentralizedtrust/splice/syncoperator/automation/ReconcileDedicatedSequencerTrafficTriggerTest.scala
@@ -0,0 +1,363 @@
+// 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.daml.metrics.api.noop.NoOpMetricsFactory
+import com.daml.nonempty.NonEmpty
+import com.digitalasset.canton.admin.api.client.data.topology.BaseResult
+import com.digitalasset.canton.concurrent.FutureSupervisor
+import com.digitalasset.canton.config.{NonNegativeFiniteDuration, ProcessingTimeout}
+import com.digitalasset.canton.config.RequireTypes.{NonNegativeLong, PositiveInt, PositiveLong}
+import com.digitalasset.canton.crypto.{Fingerprint, TestHash}
+import com.digitalasset.canton.data.CantonTimestamp
+import com.digitalasset.canton.sequencing.protocol
+import com.digitalasset.canton.time.{Clock, SimClock}
+import com.digitalasset.canton.topology.admin.grpc.TopologyStoreId
+import com.digitalasset.canton.topology.transaction.{SequencerSynchronizerState, TopologyChangeOp}
+import com.digitalasset.canton.topology.transaction.TopologyTransaction.TxHash
+import com.digitalasset.canton.topology.{
+  MediatorId,
+  Member,
+  ParticipantId,
+  SequencerId,
+  SynchronizerId,
+  UniqueIdentifier,
+}
+import com.digitalasset.canton.tracing.TraceContext
+import com.digitalasset.canton.{BaseTest, HasActorSystem, HasExecutionContext}
+import io.grpc.{Status, StatusRuntimeException}
+import org.lfdecentralizedtrust.splice.automation.{
+  TaskOutcome,
+  TaskSuccess,
+  TriggerContext,
+  TriggerEnabledSynchronization,
+}
+import org.lfdecentralizedtrust.splice.config.AutomationConfig
+import org.lfdecentralizedtrust.splice.environment.{
+  RetryProvider,
+  SequencerAdminConnection,
+  SynchronizerNodeService,
+}
+import org.lfdecentralizedtrust.splice.environment.SequencerAdminConnection.TrafficState
+import org.lfdecentralizedtrust.splice.environment.TopologyAdminConnection.{
+  TopologyResult,
+  TopologySnapshot,
+}
+import org.lfdecentralizedtrust.splice.syncoperator.SyncOperatorSynchronizerNode
+import org.lfdecentralizedtrust.splice.syncoperator.automation.ReconcileDedicatedSequencerTrafficTrigger.{
+  Task,
+  tasks,
+}
+import org.lfdecentralizedtrust.splice.syncoperator.store.SyncOperatorStore
+import org.scalatest.wordspec.AnyWordSpec
+
+import java.time.Instant
+import scala.concurrent.Future
+
+class ReconcileDedicatedSequencerTrafficTriggerTest
+    extends AnyWordSpec
+    with BaseTest
+    with HasActorSystem
+    with HasExecutionContext {
+
+  private val alice: Member = ParticipantId.tryFromProtoPrimitive("PAR::alice::default")
+  private val bob: Member = ParticipantId.tryFromProtoPrimitive("PAR::bob::default")
+  private val carol: Member = ParticipantId.tryFromProtoPrimitive("PAR::carol::default")
+  private val mediator: Member = MediatorId(UniqueIdentifier.tryCreate("mediator", "default"))
+  private val sequencer: Member = SequencerId(UniqueIdentifier.tryCreate("sequencer", "default"))
+
+  private val allowance = Some(50L)
+
+  // What the store holds as alice's purchased total.
+  private val aliceTotal = 100L
+
+  private val synchronizerId = SynchronizerId.tryFromString("dedicated::default")
+
+  "ReconcileDedicatedSequencerTrafficTrigger.tasks" should {
+
+    "grant each participant its purchased total" in {
+      val result = tasks(
+        limits = Map(alice -> 0L, bob -> 100L),
+        totals = Map(alice -> 100L, bob -> 300L),
+        allowance = None,
+      )
+      result shouldBe Seq(Task(alice, 100L, None), Task(bob, 300L, None))
+      result.map(_.target) shouldBe Seq(100L, 300L)
+    }
+
+    "leave a participant already at its purchased total alone" in {
+      tasks(
+        limits = Map(alice -> 100L),
+        totals = Map(alice -> 100L),
+        allowance = None,
+      ) shouldBe empty
+    }
+
+    "raise each participant to its purchased total plus the allowance" in {
+      val result = tasks(
+        limits = Map(alice -> 100L, bob -> 200L),
+        totals = Map(alice -> 100L, bob -> 200L),
+        allowance,
+      )
+      result shouldBe Seq(Task(alice, 100L, allowance), Task(bob, 200L, allowance))
+      result.map(_.target) shouldBe Seq(150L, 250L)
+    }
+
+    "leave a participant already at its purchased total plus the allowance alone" in {
+      tasks(
+        limits = Map(alice -> 150L),
+        totals = Map(alice -> 100L),
+        allowance,
+      ) shouldBe empty
+    }
+
+    "lower a limit a larger allowance raised to the purchased total plus a smaller one" in {
+      tasks(
+        limits = Map(alice -> 200L),
+        totals = Map(alice -> 100L),
+        allowance,
+      ).map(_.target) shouldBe Seq(150L)
+    }
+
+    "lower each limit back to the purchased total once the allowance is removed" in {
+      val result = tasks(
+        limits = Map(alice -> 150L, bob -> 200L),
+        totals = Map(alice -> 100L, bob -> 200L),
+        allowance = None,
+      )
+      result shouldBe Seq(Task(alice, 100L, None))
+      result.map(_.target) shouldBe Seq(100L)
+    }
+
+    "give a participant with no purchase on record the allowance alone" in {
+      val result = tasks(
+        limits = Map(carol -> 0L),
+        totals = Map.empty,
+        allowance,
+      )
+      result shouldBe Seq(Task(carol, 0L, allowance))
+      result.map(_.target) shouldBe Seq(50L)
+    }
+
+    "leave a participant with no purchase on record at zero without an allowance" in {
+      tasks(
+        limits = Map(carol -> 0L),
+        totals = Map.empty,
+        allowance = None,
+      ) shouldBe empty
+    }
+
+    "take back the allowance from a participant with no purchase on record once it is removed" in {
+      tasks(
+        limits = Map(carol -> 50L),
+        totals = Map.empty,
+        allowance = None,
+      ).map(_.target) shouldBe Seq(0L)
+    }
+
+    "leave a mediator's unlimited traffic alone, even with a purchase on record" in {
+      forEvery(Seq(allowance, None)) { allowance =>
+        forEvery(Seq(0L, Long.MaxValue)) { limit =>
+          tasks(
+            limits = Map(mediator -> limit),
+            totals = Map(mediator -> 10L),
+            allowance,
+          ) shouldBe empty
+        }
+      }
+    }
+
+    "leave a sequencer alone" in {
+      tasks(
+        limits = Map(sequencer -> 0L),
+        totals = Map(sequencer -> 10L),
+        allowance,
+      ) shouldBe empty
+    }
+
+    "wait with a purchase for a member the sequencer has no traffic state for" in {
+      tasks(
+        limits = Map(alice -> 100L),
+        totals = Map(alice -> 100L, carol -> 10L),
+        allowance = None,
+      ) shouldBe empty
+    }
+
+    "list the participants in a fixed order" in {
+      tasks(
+        limits = Map(carol -> 0L, mediator -> 0L, alice -> 0L, bob -> 0L),
+        totals = Map(carol -> 1L, alice -> 1L, bob -> 1L),
+        allowance,
+      ).map(_.member) shouldBe Seq(alice, bob, carol)
+    }
+  }
+
+  "ReconcileDedicatedSequencerTrafficTrigger.Task" should {
+
+    "cap a target the allowance would take past the largest limit" in {
+      Task(alice, 100L, Some(Long.MaxValue)).target shouldBe Long.MaxValue
+      Task(alice, Long.MaxValue, allowance).target shouldBe Long.MaxValue
+    }
+
+    "leave a participant at the largest limit alone once its target is capped" in {
+      tasks(
+        limits = Map(alice -> Long.MaxValue),
+        totals = Map(alice -> 100L),
+        allowance = Some(Long.MaxValue),
+      ) shouldBe empty
+    }
+  }
+
+  "ReconcileDedicatedSequencerTrafficTrigger" should {
+
+    val task = Task(alice, aliceTotal, allowance)
+    // Her purchased total plus the allowance.
+    val target = 150L
+
+    "set a participant's limit to its target" in {
+      val f = new Fixture(limit = 0L)
+      f.grantReturns(Future.unit)
+      f.trigger.complete(task).futureValue shouldBe a[TaskSuccess]
+      f.verifyGrant(target)
+    }
+
+    "set the target from the participant's total as it is now" in {
+      // Retrieved before a purchase of the whole total was ingested.
+      val f = new Fixture(limit = 0L)
+      f.grantReturns(Future.unit)
+      f.trigger.complete(task.copy(total = 0L)).futureValue shouldBe a[TaskSuccess]
+      f.verifyGrant(target)
+    }
+
+    "fail a grant that fails, so that it is retried" in {
+      val f = new Fixture(limit = 0L)
+      f.grantReturns(Future.failed(Status.DEADLINE_EXCEEDED.asRuntimeException()))
+      failureCode(f.trigger.complete(task)) shouldBe Status.Code.DEADLINE_EXCEEDED
+    }
+
+    "treat a task as stale once the participant is at its target" in {
+      new Fixture(limit = target).trigger.stale(task).futureValue shouldBe true
+    }
+
+    "treat a task as stale once the participant's current total puts it at its target" in {
+      // Retrieved before a purchase of the whole total, which has been granted since.
+      new Fixture(limit = target).trigger
+        .stale(task.copy(total = 0L))
+        .futureValue shouldBe true
+    }
+
+    "treat a task as stale once the participant has no traffic state" in {
+      new Fixture(limit = 0L, hasTrafficState = false).trigger
+        .stale(task)
+        .futureValue shouldBe true
+    }
+
+    "not treat a task as stale while the participant is not at its target" in {
+      new Fixture(limit = 0L).trigger.stale(task).futureValue shouldBe false
+    }
+  }
+
+  private lazy val clock = new SimClock(loggerFactory = loggerFactory)
+
+  private lazy val triggerContext = TriggerContext(
+    AutomationConfig(),
+    clock,
+    clock,
+    TriggerEnabledSynchronization.Noop,
+    RetryProvider(loggerFactory, ProcessingTimeout(), FutureSupervisor.Noop, NoOpMetricsFactory),
+    loggerFactory,
+    NoOpMetricsFactory,
+  )
+
+  private def failureCode(outcome: Future[TaskOutcome]): Status.Code =
+    inside(outcome.failed.futureValue) { case e: StatusRuntimeException => e.getStatus.getCode }
+
+  /** The trigger on a mocked sequencer, store and node service, with alice at `limit`. */
+  private class Fixture(limit: Long, hasTrafficState: Boolean = true) {
+    val connection: SequencerAdminConnection = mock[SequencerAdminConnection]
+    private val store = mock[SyncOperatorStore]
+    private val nodeService = mock[SynchronizerNodeService[SyncOperatorSynchronizerNode]]
+
+    when(nodeService.sequencerAdminConnection()(any[TraceContext]))
+      .thenReturn(Future.successful(connection))
+    when(store.getTotalPurchasedMemberTraffic(eqTo(alice))(any[TraceContext]))
+      .thenReturn(Future.successful(aliceTotal))
+    private val trafficState = TrafficState(
+      alice,
+      protocol.TrafficState
+        .empty(CantonTimestamp.Epoch)
+        .copy(
+          extraTrafficPurchased = NonNegativeLong.tryCreate(limit),
+          serial = Some(PositiveInt.one),
+        ),
+    )
+    when(connection.getSequencerTrafficControlState(eqTo(alice))(any[TraceContext]))
+      .thenReturn(
+        if (hasTrafficState) Future.successful(trafficState)
+        else Future.failed(Status.NOT_FOUND.asRuntimeException())
+      )
+    when(connection.lookupSequencerTrafficControlState(eqTo(alice))(any[TraceContext]))
+      .thenReturn(Future.successful(Option.when(hasTrafficState)(trafficState)))
+    when(connection.getSequencerSynchronizerState(any[TopologySnapshot])(any[TraceContext]))
+      .thenReturn(Future.successful(sequencerState(threshold = 1)))
+
+    val trigger = new TestTrigger(store, nodeService)
+
+    def grantReturns(result: Future[Unit]): Unit =
+      when(
+        connection.setSequencerTrafficControlState(
+          any[TrafficState],
+          any[TopologyResult[SequencerSynchronizerState]],
+          any[NonNegativeLong],
+          any[Clock],
+          any[NonNegativeFiniteDuration],
+        )(any[TraceContext])
+      ).thenReturn(result)
+
+    def verifyGrant(target: Long): Unit =
+      verify(connection).setSequencerTrafficControlState(
+        any[TrafficState],
+        any[TopologyResult[SequencerSynchronizerState]],
+        eqTo(NonNegativeLong.tryCreate(target)),
+        any[Clock],
+        any[NonNegativeFiniteDuration],
+      )(any[TraceContext])
+  }
+
+  private class TestTrigger(
+      store: SyncOperatorStore,
+      nodeService: SynchronizerNodeService[SyncOperatorSynchronizerNode],
+  ) extends ReconcileDedicatedSequencerTrafficTrigger(
+        triggerContext,
+        store,
+        nodeService,
+        NonNegativeFiniteDuration.ofSeconds(10),
+        allowance.map(PositiveLong.tryCreate),
+      ) {
+    def complete(task: Task): Future[TaskOutcome] = completeTask(task)
+    def stale(task: Task): Future[Boolean] = isStaleTask(task)
+  }
+
+  private def sequencerState(threshold: Int): TopologyResult[SequencerSynchronizerState] =
+    TopologyResult(
+      BaseResult(
+        storeId = TopologyStoreId.Authorized,
+        validFrom = Instant.EPOCH,
+        validUntil = None,
+        sequenced = Instant.EPOCH,
+        operation = TopologyChangeOp.Replace,
+        transactionHash = TxHash(TestHash.digest("sequencer-state")),
+        serial = PositiveInt.one,
+        signedBy = NonEmpty(Seq, Fingerprint.tryFromString("default")),
+      ),
+      SequencerSynchronizerState
+        .create(
+          synchronizerId,
+          PositiveInt.tryCreate(threshold),
+          (1 to threshold).map(i => SequencerId(UniqueIdentifier.tryCreate(s"seq$i", "default"))),
+          Seq.empty,
+        )
+        .value,
+    )
+}
diff --git a/test-full-class-names-non-integration.log b/test-full-class-names-non-integration.log
index 483884d369b..392ca705358 100644
--- a/test-full-class-names-non-integration.log
+++ b/test-full-class-names-non-integration.log
@@ -59,6 +59,7 @@ org.lfdecentralizedtrust.splice.sv.cometbft.CometBftRequestSignerTest
 org.lfdecentralizedtrust.splice.sv.onboarding.SequencerBftPeerReconcilerSpec
 org.lfdecentralizedtrust.splice.sv.util.SvOnboardingTokenTest
 org.lfdecentralizedtrust.splice.sv.util.WeightDistributionForSvTest
+org.lfdecentralizedtrust.splice.syncoperator.automation.ReconcileDedicatedSequencerTrafficTriggerTest
 org.lfdecentralizedtrust.splice.unit.http.HttpClientProxyTest
 org.lfdecentralizedtrust.splice.unit.store.db.AcsJdbcTypesTest
 org.lfdecentralizedtrust.splice.util.AmuletConfigScheduleTest