From 3961a7fea80a05c6eccff22bee5720e065d57c03 Mon Sep 17 00:00:00 2001 From: sadiq1971 Date: Wed, 12 Aug 2026 00:38:21 +0600 Subject: [PATCH 1/5] refactor(scala-sv): extract reusable member-traffic reconciliation trigger [ci] Move the reconciliation logic into an abstract trigger in apps/common, parameterized by the target synchronizer and its sequencer admin connection, so the Sync Operator Node can reconcile a dedicated synchronizer with the same code. The SV subclass keeps its name and package so its canonical name, metrics and paused-trigger key are unchanged. Signed-off-by: sadiq1971 --- ...cerLimitWithMemberTrafficTriggerBase.scala | 204 ++++++++++++++++++ .../automation/SvDsoAutomationService.scala | 1 + ...quencerLimitWithMemberTrafficTrigger.scala | 190 ++++------------ 3 files changed, 241 insertions(+), 154 deletions(-) create mode 100644 apps/common/src/main/scala/org/lfdecentralizedtrust/splice/automation/ReconcileSequencerLimitWithMemberTrafficTriggerBase.scala 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 new file mode 100644 index 0000000000..e23a5d342b --- /dev/null +++ b/apps/common/src/main/scala/org/lfdecentralizedtrust/splice/automation/ReconcileSequencerLimitWithMemberTrafficTriggerBase.scala @@ -0,0 +1,204 @@ +// Copyright (c) 2024 Digital Asset (Switzerland) GmbH and/or its affiliates. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +package org.lfdecentralizedtrust.splice.automation + +import com.digitalasset.canton.config.NonNegativeFiniteDuration +import com.digitalasset.canton.config.RequireTypes.NonNegativeLong +import com.digitalasset.canton.topology.{Member, SynchronizerId} +import com.digitalasset.canton.tracing.TraceContext +import io.grpc.Status +import io.opentelemetry.api.trace.Tracer +import org.apache.pekko.stream.Materializer +import org.lfdecentralizedtrust.splice.codegen.java.splice +import org.lfdecentralizedtrust.splice.environment.SequencerAdminConnection +import org.lfdecentralizedtrust.splice.environment.TopologyAdminConnection.TopologySnapshot +import org.lfdecentralizedtrust.splice.store.AppStore +import org.lfdecentralizedtrust.splice.util.AssignedContract + +import scala.concurrent.{ExecutionContext, Future} + +/** Grants the traffic purchased via `MemberTraffic` contracts on the sequencer of a single + * synchronizer. + * + * Traffic can be purchased for any synchronizer, so a node observes `MemberTraffic` contracts + * naming synchronizers it does not serve. Each instance of this trigger owns exactly one + * synchronizer, given by [[targetSynchronizerId]], and skips everything else; the operator of + * another synchronizer grants those purchases on its own sequencer. + * + * This trigger currently relies on enough nodes 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. + * + * TODO(tech-debt): remove this constraint by ensuring that we regularly submit set-traffic-balance + * requests for ALL members. + */ +abstract class ReconcileSequencerLimitWithMemberTrafficTriggerBase( + store: AppStore, + trafficBalanceReconciliationDelay: NonNegativeFiniteDuration, +)(implicit + ec: ExecutionContext, + mat: Materializer, + tracer: Tracer, +) extends OnAssignedContractTrigger.Template[ + splice.decentralizedsynchronizer.MemberTraffic.ContractId, + splice.decentralizedsynchronizer.MemberTraffic, + ]( + store, + splice.decentralizedsynchronizer.MemberTraffic.COMPANION, + ) { + + /** The synchronizer this trigger reconciles. `MemberTraffic` contracts for any other + * synchronizer are skipped. + */ + protected def targetSynchronizerId()(implicit tc: TraceContext): Future[SynchronizerId] + + /** Admin connection to the sequencer serving [[targetSynchronizerId]]. */ + protected def sequencerAdminConnection()(implicit + tc: TraceContext + ): Future[SequencerAdminConnection] + + /** Total traffic purchased for `memberId` on `synchronizerId`. */ + protected def getTotalPurchasedMemberTraffic(memberId: Member, synchronizerId: SynchronizerId)( + implicit tc: TraceContext + ): Future[Long] + + /** The traffic already consumed by `memberId` before this trigger took over its reconciliation, + * added to the purchased total when setting the limit. `None` skips the member entirely, which + * is how members that are granted unlimited traffic are excluded. + */ + protected def trafficLimitOffset(memberId: Member, synchronizerId: SynchronizerId)(implicit + tc: TraceContext + ): Future[Option[Long]] + + override def completeTask( + memberTraffic: AssignedContract[ + splice.decentralizedsynchronizer.MemberTraffic.ContractId, + splice.decentralizedsynchronizer.MemberTraffic, + ] + )(implicit tc: TraceContext): Future[TaskOutcome] = { + Member + .fromProtoPrimitive_(memberTraffic.payload.memberId) + .fold( + err => + // Skip contracts with invalid member ids + Future.successful(TaskSuccess(s"Skipping MemberTraffic with invalid memberId: ${err}")), + memberId => + SynchronizerId + .fromString(memberTraffic.payload.synchronizerId) + .fold( + err => + // Skip contracts with invalid synchronizer ids + Future.successful( + TaskSuccess(s"Skipping MemberTraffic with invalid synchronizerId: ${err}") + ), + synchronizerId => + targetSynchronizerId().flatMap { + case target if target != synchronizerId => + Future.successful( + TaskSuccess( + s"Skipping MemberTraffic contract for synchronizer $synchronizerId, " + + s"this trigger reconciles $target" + ) + ) + case target => + reconcileOnTargetSequencer(memberId, target) + }, + ), + ) + } + + private def reconcileOnTargetSequencer(memberId: Member, synchronizerId: SynchronizerId)(implicit + tc: TraceContext + ): Future[TaskOutcome] = + sequencerAdminConnection().flatMap { sequencerAdminConnection => + sequencerAdminConnection.getStatus + .map(_.successOption.map(_.synchronizerId)) + .flatMap { + case None => + Future.failed( + Status.FAILED_PRECONDITION + .withDescription("Sequencer is not yet initialized") + .asRuntimeException() + ) + case Some(sequencerSynchronizerId) + if sequencerSynchronizerId.logical != synchronizerId => + // The connection does not serve the synchronizer this trigger was configured for, + // so granting traffic here would credit the wrong sequencer. + Future.failed( + Status.INTERNAL + .withDescription( + s"The sequencer admin connection serves ${sequencerSynchronizerId.logical}, " + + s"but this trigger reconciles $synchronizerId" + ) + .asRuntimeException() + ) + case _ => + trafficLimitOffset(memberId, synchronizerId).flatMap { + case None => + Future.successful( + TaskSuccess(s"Skipping MemberTraffic contract for member $memberId") + ) + case Some(offset) => + reconcileExtraTrafficLimitForMember( + memberId, + synchronizerId, + offset, + sequencerAdminConnection, + ) + } + } + } + + private def reconcileExtraTrafficLimitForMember( + memberId: Member, + synchronizerId: SynchronizerId, + trafficLimitOffset: Long, + sequencerAdminConnection: SequencerAdminConnection, + )(implicit tc: TraceContext): Future[TaskSuccess] = { + sequencerAdminConnection.lookupSequencerTrafficControlState(memberId).flatMap { + case None => + Future.successful( + TaskSuccess( + s"No traffic state found for member $memberId. It is likely that the member has been disabled as it was lagging behind and prevented sequencer pruning." + ) + ) + case Some(trafficState) => + for { + // Compute new extra traffic limit + totalPurchasedTraffic <- getTotalPurchasedMemberTraffic(memberId, synchronizerId) + newExtraTrafficLimit = NonNegativeLong + .tryCreate(trafficLimitOffset + totalPurchasedTraffic) + + // Get current effective sequencer domain state + sequencerSynchronizerState <- sequencerAdminConnection + .getSequencerSynchronizerState(topologySnapshot = TopologySnapshot.Effective) + currentExtraTrafficLimit = trafficState.extraTrafficLimit + + // Compare and reconcile old and new limits + taskOutcome <- + if (currentExtraTrafficLimit < newExtraTrafficLimit) { + sequencerAdminConnection + .setSequencerTrafficControlState( + trafficState, + sequencerSynchronizerState, + newExtraTrafficLimit, + context.pollingClock, + trafficBalanceReconciliationDelay, + ) + .map(_ => + TaskSuccess( + s"Updated extra traffic limit for member ${memberId} from ${currentExtraTrafficLimit} to ${newExtraTrafficLimit}" + ) + ) + } else { + Future( + TaskSuccess( + s"Skipping since traffic limit is already up to date (previous limit = ${currentExtraTrafficLimit}, new limit = ${newExtraTrafficLimit})." + ) + ) + } + } yield taskOutcome + } + } +} diff --git a/apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/automation/SvDsoAutomationService.scala b/apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/automation/SvDsoAutomationService.scala index b67dafbebb..ee2a222ebe 100644 --- a/apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/automation/SvDsoAutomationService.scala +++ b/apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/automation/SvDsoAutomationService.scala @@ -470,6 +470,7 @@ class SvDsoAutomationService( triggerContext, dsoStore, synchronizerNodeService, + synchronizerId, config.trafficBalanceReconciliationDelay, ) ) diff --git a/apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/automation/singlesv/ReconcileSequencerLimitWithMemberTrafficTrigger.scala b/apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/automation/singlesv/ReconcileSequencerLimitWithMemberTrafficTrigger.scala index 90cb1a442d..695ea77888 100644 --- a/apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/automation/singlesv/ReconcileSequencerLimitWithMemberTrafficTrigger.scala +++ b/apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/automation/singlesv/ReconcileSequencerLimitWithMemberTrafficTrigger.scala @@ -4,193 +4,75 @@ package org.lfdecentralizedtrust.splice.sv.automation.singlesv import com.digitalasset.canton.config.NonNegativeFiniteDuration -import com.digitalasset.canton.config.RequireTypes.NonNegativeLong import com.digitalasset.canton.topology.{Member, SynchronizerId} import com.digitalasset.canton.tracing.TraceContext -import io.grpc.Status import io.opentelemetry.api.trace.Tracer import org.apache.pekko.stream.Materializer import org.lfdecentralizedtrust.splice.automation.{ - OnAssignedContractTrigger, - TaskOutcome, - TaskSuccess, + ReconcileSequencerLimitWithMemberTrafficTriggerBase, TriggerContext, } -import org.lfdecentralizedtrust.splice.codegen.java.splice import org.lfdecentralizedtrust.splice.environment.{ SequencerAdminConnection, SynchronizerNodeService, } -import org.lfdecentralizedtrust.splice.environment.TopologyAdminConnection.TopologySnapshot import org.lfdecentralizedtrust.splice.sv.store.SvDsoStore import org.lfdecentralizedtrust.splice.sv.LocalSynchronizerNode -import org.lfdecentralizedtrust.splice.util.AssignedContract import scala.concurrent.{ExecutionContext, Future} import scala.jdk.CollectionConverters.* -/** 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. - * - * TODO(tech-debt): remove this constraint by ensuring that we regularly submit set-traffic-balance requests for ALL members. +/** Reconciles the traffic purchased on the decentralized synchronizer with this SV's sequencer. + * Purchases for other synchronizers are granted by their own operators, so they are skipped here. */ class ReconcileSequencerLimitWithMemberTrafficTrigger( override protected val context: TriggerContext, store: SvDsoStore, synchronizerNodeService: SynchronizerNodeService[LocalSynchronizerNode], + decentralizedSynchronizerId: SynchronizerId, trafficBalanceReconciliationDelay: NonNegativeFiniteDuration, )(implicit ec: ExecutionContext, mat: Materializer, tracer: Tracer, -) extends OnAssignedContractTrigger.Template[ - splice.decentralizedsynchronizer.MemberTraffic.ContractId, - splice.decentralizedsynchronizer.MemberTraffic, - ]( +) extends ReconcileSequencerLimitWithMemberTrafficTriggerBase( store, - splice.decentralizedsynchronizer.MemberTraffic.COMPANION, + trafficBalanceReconciliationDelay, ) { - override def completeTask( - memberTraffic: AssignedContract[ - splice.decentralizedsynchronizer.MemberTraffic.ContractId, - splice.decentralizedsynchronizer.MemberTraffic, - ] - )(implicit tc: TraceContext): Future[TaskOutcome] = { - Member - .fromProtoPrimitive_(memberTraffic.payload.memberId) - .fold( - err => { - // Skip contracts with invalid member ids - Future.successful(TaskSuccess(s"Skipping MemberTraffic with invalid memberId: ${err}")) - }, - memberId => - SynchronizerId - .fromString(memberTraffic.payload.synchronizerId) - .fold( - err => { - // Unlike a foreign synchronizer id, an unparseable one means corrupt data and - // should never be routine, so it is worth an alarm as well as a skip. - logger.warn(s"Skipping MemberTraffic with unparseable synchronizerId: ${err}") - Future.successful( - TaskSuccess(s"Skipping MemberTraffic with invalid synchronizerId: ${err}") - ) - }, - synchronizerId => - synchronizerNodeService.sequencerAdminConnection().flatMap { - sequencerAdminConnection => - sequencerAdminConnection.getStatus - .map(_.successOption.map(_.synchronizerId)) - .flatMap { - case None => - Future.failed( - Status.FAILED_PRECONDITION - .withDescription("Sequencer is not yet initialized") - .asRuntimeException() - ) - case Some(sequencerSynchronizerId) - if sequencerSynchronizerId.logical != synchronizerId => - // Traffic can be purchased for any registered synchronizer, so we - // observe MemberTraffic contracts that this sequencer does not serve. - // They are granted by the operator of the synchronizer they name, on - // that synchronizer's own sequencer, so skip them here rather than - // failing the trigger. - Future.successful( - TaskSuccess( - s"Skipping MemberTraffic contract for synchronizer " + - s"$synchronizerId, this sequencer serves " + - s"${sequencerSynchronizerId.logical}" - ) - ) - case _ => - store - .getDsoRulesWithSvNodeStates() - .flatMap(rulesAndStates => { - if ( - rulesAndStates - .activeSvParticipantAndMediatorIds(synchronizerId) - .contains(memberId) - ) { - // SVs are granted unlimited traffic and do not need to purchase - // it via MemberTraffic contracts. While the top-up trigger for SV - // validators is disabled by default, we also explicitly ignore SV - // related MemberTraffic contracts here as a safeguard for the case - // of 3rd party top-ups of SV nodes or an SV validator - // misconfiguration that changes the defaults. - Future - .successful( - TaskSuccess( - s"Skipping MemberTraffic contract for SV node $memberId" - ) - ) - } else { - val trafficLimitOffset = - rulesAndStates.dsoRules.payload.initialTrafficState.asScala - .get(memberId.toProtoPrimitive) - .fold(0L)(_.consumedTraffic) - reconcileExtraTrafficLimitForMember( - memberId, - synchronizerId, - trafficLimitOffset, - sequencerAdminConnection, - ) - } - }) - } - }, - ), - ) - } + override protected def targetSynchronizerId()(implicit + tc: TraceContext + ): Future[SynchronizerId] = + Future.successful(decentralizedSynchronizerId) - private def reconcileExtraTrafficLimitForMember( + override protected def sequencerAdminConnection()(implicit + tc: TraceContext + ): Future[SequencerAdminConnection] = + synchronizerNodeService.sequencerAdminConnection() + + override protected def getTotalPurchasedMemberTraffic( memberId: Member, synchronizerId: SynchronizerId, - trafficLimitOffset: Long, - sequencerAdminConnection: SequencerAdminConnection, - )(implicit tc: TraceContext): Future[TaskSuccess] = { - sequencerAdminConnection.lookupSequencerTrafficControlState(memberId).flatMap { - case None => - Future.successful( - TaskSuccess( - s"No traffic state found for member $memberId. It is likely that the member has been disabled as it was lagging behind and prevented sequencer pruning." - ) - ) - case Some(trafficState) => - for { - // Compute new extra traffic limit - totalPurchasedTraffic <- store.getTotalPurchasedMemberTraffic(memberId, synchronizerId) - newExtraTrafficLimit = NonNegativeLong - .tryCreate(trafficLimitOffset + totalPurchasedTraffic) + )(implicit tc: TraceContext): Future[Long] = + store.getTotalPurchasedMemberTraffic(memberId, synchronizerId) - // Get current effective sequencer domain state - sequencerSynchronizerState <- sequencerAdminConnection - .getSequencerSynchronizerState(topologySnapshot = TopologySnapshot.Effective) - currentExtraTrafficLimit = trafficState.extraTrafficLimit - - // Compare and reconcile old and new limits - taskOutcome <- - if (currentExtraTrafficLimit < newExtraTrafficLimit) { - sequencerAdminConnection - .setSequencerTrafficControlState( - trafficState, - sequencerSynchronizerState, - newExtraTrafficLimit, - context.pollingClock, - trafficBalanceReconciliationDelay, - ) - .map(_ => - TaskSuccess( - s"Updated extra traffic limit for member ${memberId} from ${currentExtraTrafficLimit} to ${newExtraTrafficLimit}" - ) - ) - } else { - Future( - TaskSuccess( - s"Skipping since traffic limit is already up to date (previous limit = ${currentExtraTrafficLimit}, new limit = ${newExtraTrafficLimit})." - ) - ) - } - } yield taskOutcome + override protected def trafficLimitOffset(memberId: Member, synchronizerId: SynchronizerId)( + implicit tc: TraceContext + ): Future[Option[Long]] = + store.getDsoRulesWithSvNodeStates().map { rulesAndStates => + if (rulesAndStates.activeSvParticipantAndMediatorIds(synchronizerId).contains(memberId)) { + // SVs are granted unlimited traffic and do not need to purchase it via MemberTraffic + // contracts. While the top-up trigger for SV validators is disabled by default, we also + // explicitly ignore SV related MemberTraffic contracts here as a safeguard for the case of + // 3rd party top-ups of SV nodes or an SV validator misconfiguration that changes the + // defaults. + None + } else { + Some( + rulesAndStates.dsoRules.payload.initialTrafficState.asScala + .get(memberId.toProtoPrimitive) + .fold(0L)(_.consumedTraffic) + ) + } } - } } From 576bc8d88c909f0189c80c5a1fa1b428cf9bcbf8 Mon Sep 17 00:00:00 2001 From: sadiq1971 Date: Wed, 12 Aug 2026 01:30:54 +0600 Subject: [PATCH 2/5] refactor(scala-sv): address review on the reusable traffic trigger [ci] - report a target/connection mismatch as FAILED_PRECONDITION so it is retryable instead of dropping the purchase with an ERROR - warn once when the configured target is not served by the sequencer, which a per-contract check cannot see because every contract skips first - carry the skip reason through trafficLimitOffset instead of overloading Option - pin the target to a stable value and drop the redundant synchronizer id from the store hooks - skip members before opening the sequencer connection Signed-off-by: sadiq1971 --- ...cerLimitWithMemberTrafficTriggerBase.scala | 142 +++++++++++------- ...quencerLimitWithMemberTrafficTrigger.scala | 30 ++-- 2 files changed, 98 insertions(+), 74 deletions(-) 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 e23a5d342b..f1374338d2 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 @@ -16,6 +16,7 @@ import org.lfdecentralizedtrust.splice.environment.TopologyAdminConnection.Topol import org.lfdecentralizedtrust.splice.store.AppStore import org.lfdecentralizedtrust.splice.util.AssignedContract +import java.util.concurrent.atomic.AtomicBoolean import scala.concurrent.{ExecutionContext, Future} /** Grants the traffic purchased via `MemberTraffic` contracts on the sequencer of a single @@ -26,6 +27,15 @@ import scala.concurrent.{ExecutionContext, Future} * synchronizer, given by [[targetSynchronizerId]], and skips everything else; the operator of * another synchronizer grants those purchases on its own sequencer. * + * [[targetSynchronizerId]] is a stable value for the lifetime of the trigger, so a node that + * changes the logical synchronizer it serves has to be restarted. Note that the sibling + * `SvOnboardingUnlimitedTrafficTrigger` instead re-resolves the active synchronizer from + * `AmuletConfigSchedule` on every run, so the two source the same concept differently. + * + * Purchases are matched on the synchronizer id alone. Whether a purchase recorded against a + * different migration id of the same synchronizer is visible at all is decided upstream, by the + * ingestion filter of the store backing this trigger. + * * This trigger currently relies on enough nodes 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. @@ -51,25 +61,28 @@ abstract class ReconcileSequencerLimitWithMemberTrafficTriggerBase( /** The synchronizer this trigger reconciles. `MemberTraffic` contracts for any other * synchronizer are skipped. */ - protected def targetSynchronizerId()(implicit tc: TraceContext): Future[SynchronizerId] + protected def targetSynchronizerId: SynchronizerId /** Admin connection to the sequencer serving [[targetSynchronizerId]]. */ protected def sequencerAdminConnection()(implicit tc: TraceContext ): Future[SequencerAdminConnection] - /** Total traffic purchased for `memberId` on `synchronizerId`. */ - protected def getTotalPurchasedMemberTraffic(memberId: Member, synchronizerId: SynchronizerId)( - implicit tc: TraceContext + /** Total traffic purchased for `memberId` on [[targetSynchronizerId]]. */ + protected def getTotalPurchasedMemberTraffic(memberId: Member)(implicit + tc: TraceContext ): Future[Long] /** The traffic already consumed by `memberId` before this trigger took over its reconciliation, - * added to the purchased total when setting the limit. `None` skips the member entirely, which - * is how members that are granted unlimited traffic are excluded. + * added to the purchased total when setting the limit. A `Left` skips the member and carries the + * reason, which is how members that are granted unlimited traffic are excluded. */ - protected def trafficLimitOffset(memberId: Member, synchronizerId: SynchronizerId)(implicit + protected def trafficLimitOffset(memberId: Member)(implicit tc: TraceContext - ): Future[Option[Long]] + ): Future[Either[String, Long]] + + /** Guards the one-off check in [[warnOnceIfTargetNotServed]]. */ + private val targetChecked = new AtomicBoolean(false) override def completeTask( memberTraffic: AssignedContract[ @@ -93,67 +106,82 @@ abstract class ReconcileSequencerLimitWithMemberTrafficTriggerBase( TaskSuccess(s"Skipping MemberTraffic with invalid synchronizerId: ${err}") ), synchronizerId => - targetSynchronizerId().flatMap { - case target if target != synchronizerId => - Future.successful( - TaskSuccess( - s"Skipping MemberTraffic contract for synchronizer $synchronizerId, " + - s"this trigger reconciles $target" - ) + if (synchronizerId != targetSynchronizerId) { + // A misconfigured target makes every contract land here, which would otherwise + // leave the trigger silently granting nothing, so check the target itself once. + warnOnceIfTargetNotServed().map { _ => + TaskSuccess( + s"Skipping MemberTraffic contract for synchronizer $synchronizerId, " + + s"this trigger reconciles $targetSynchronizerId" ) - case target => - reconcileOnTargetSequencer(memberId, target) + } + } else { + reconcileMember(memberId) }, ), ) } - private def reconcileOnTargetSequencer(memberId: Member, synchronizerId: SynchronizerId)(implicit - tc: TraceContext - ): Future[TaskOutcome] = - sequencerAdminConnection().flatMap { sequencerAdminConnection => - sequencerAdminConnection.getStatus - .map(_.successOption.map(_.synchronizerId)) - .flatMap { - case None => - Future.failed( - Status.FAILED_PRECONDITION - .withDescription("Sequencer is not yet initialized") - .asRuntimeException() - ) - case Some(sequencerSynchronizerId) - if sequencerSynchronizerId.logical != synchronizerId => - // The connection does not serve the synchronizer this trigger was configured for, - // so granting traffic here would credit the wrong sequencer. - Future.failed( - Status.INTERNAL - .withDescription( - s"The sequencer admin connection serves ${sequencerSynchronizerId.logical}, " + - s"but this trigger reconciles $synchronizerId" - ) - .asRuntimeException() - ) - case _ => - trafficLimitOffset(memberId, synchronizerId).flatMap { + private def reconcileMember(memberId: Member)(implicit tc: TraceContext): Future[TaskOutcome] = + trafficLimitOffset(memberId).flatMap { + case Left(reason) => + Future.successful(TaskSuccess(s"Skipping MemberTraffic contract for $memberId: $reason")) + case Right(offset) => + sequencerAdminConnection().flatMap { sequencerAdminConnection => + sequencerAdminConnection.getStatus + .map(_.successOption.map(_.synchronizerId)) + .flatMap { case None => - Future.successful( - TaskSuccess(s"Skipping MemberTraffic contract for member $memberId") + Future.failed( + Status.FAILED_PRECONDITION + .withDescription("Sequencer is not yet initialized") + .asRuntimeException() ) - case Some(offset) => - reconcileExtraTrafficLimitForMember( - memberId, - synchronizerId, - offset, - sequencerAdminConnection, + case Some(sequencerSynchronizerId) + if sequencerSynchronizerId.logical != targetSynchronizerId => + // Granting here would credit a sequencer of a different synchronizer. Reported as + // retryable so that a connection that is still switching over recovers on its own + // instead of dropping the purchase. + Future.failed( + Status.FAILED_PRECONDITION + .withDescription( + s"The sequencer admin connection serves ${sequencerSynchronizerId.logical}, " + + s"but this trigger reconciles $targetSynchronizerId" + ) + .asRuntimeException() ) + case _ => + reconcileExtraTrafficLimitForMember(memberId, offset, sequencerAdminConnection) + } + } + } + + /** Logs at most once if the sequencer we would grant on does not serve [[targetSynchronizerId]]. + * A sequencer that is not initialized yet leaves the check pending for a later task. + */ + private def warnOnceIfTargetNotServed()(implicit tc: TraceContext): Future[Unit] = + if (targetChecked.get()) { + Future.unit + } else { + sequencerAdminConnection() + .flatMap(_.getStatus) + .map(_.successOption.map(_.synchronizerId)) + .map { + case Some(sequencerSynchronizerId) => + if (sequencerSynchronizerId.logical != targetSynchronizerId) { + logger.warn( + s"This trigger reconciles $targetSynchronizerId, but its sequencer serves " + + s"${sequencerSynchronizerId.logical}, so no traffic will ever be granted" + ) } + targetChecked.set(true) + case None => () } } private def reconcileExtraTrafficLimitForMember( memberId: Member, - synchronizerId: SynchronizerId, - trafficLimitOffset: Long, + offset: Long, sequencerAdminConnection: SequencerAdminConnection, )(implicit tc: TraceContext): Future[TaskSuccess] = { sequencerAdminConnection.lookupSequencerTrafficControlState(memberId).flatMap { @@ -166,9 +194,9 @@ abstract class ReconcileSequencerLimitWithMemberTrafficTriggerBase( case Some(trafficState) => for { // Compute new extra traffic limit - totalPurchasedTraffic <- getTotalPurchasedMemberTraffic(memberId, synchronizerId) + totalPurchasedTraffic <- getTotalPurchasedMemberTraffic(memberId) newExtraTrafficLimit = NonNegativeLong - .tryCreate(trafficLimitOffset + totalPurchasedTraffic) + .tryCreate(offset + totalPurchasedTraffic) // Get current effective sequencer domain state sequencerSynchronizerState <- sequencerAdminConnection @@ -192,7 +220,7 @@ abstract class ReconcileSequencerLimitWithMemberTrafficTriggerBase( ) ) } else { - Future( + Future.successful( TaskSuccess( s"Skipping since traffic limit is already up to date (previous limit = ${currentExtraTrafficLimit}, new limit = ${newExtraTrafficLimit})." ) diff --git a/apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/automation/singlesv/ReconcileSequencerLimitWithMemberTrafficTrigger.scala b/apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/automation/singlesv/ReconcileSequencerLimitWithMemberTrafficTrigger.scala index 695ea77888..87cf3728ec 100644 --- a/apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/automation/singlesv/ReconcileSequencerLimitWithMemberTrafficTrigger.scala +++ b/apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/automation/singlesv/ReconcileSequencerLimitWithMemberTrafficTrigger.scala @@ -29,7 +29,7 @@ class ReconcileSequencerLimitWithMemberTrafficTrigger( override protected val context: TriggerContext, store: SvDsoStore, synchronizerNodeService: SynchronizerNodeService[LocalSynchronizerNode], - decentralizedSynchronizerId: SynchronizerId, + override protected val targetSynchronizerId: SynchronizerId, trafficBalanceReconciliationDelay: NonNegativeFiniteDuration, )(implicit ec: ExecutionContext, @@ -40,35 +40,31 @@ class ReconcileSequencerLimitWithMemberTrafficTrigger( trafficBalanceReconciliationDelay, ) { - override protected def targetSynchronizerId()(implicit - tc: TraceContext - ): Future[SynchronizerId] = - Future.successful(decentralizedSynchronizerId) - override protected def sequencerAdminConnection()(implicit tc: TraceContext ): Future[SequencerAdminConnection] = synchronizerNodeService.sequencerAdminConnection() - override protected def getTotalPurchasedMemberTraffic( - memberId: Member, - synchronizerId: SynchronizerId, - )(implicit tc: TraceContext): Future[Long] = - store.getTotalPurchasedMemberTraffic(memberId, synchronizerId) + override protected def getTotalPurchasedMemberTraffic(memberId: Member)(implicit + tc: TraceContext + ): Future[Long] = + store.getTotalPurchasedMemberTraffic(memberId, targetSynchronizerId) - override protected def trafficLimitOffset(memberId: Member, synchronizerId: SynchronizerId)( - implicit tc: TraceContext - ): Future[Option[Long]] = + override protected def trafficLimitOffset(memberId: Member)(implicit + tc: TraceContext + ): Future[Either[String, Long]] = store.getDsoRulesWithSvNodeStates().map { rulesAndStates => - if (rulesAndStates.activeSvParticipantAndMediatorIds(synchronizerId).contains(memberId)) { + if ( + rulesAndStates.activeSvParticipantAndMediatorIds(targetSynchronizerId).contains(memberId) + ) { // SVs are granted unlimited traffic and do not need to purchase it via MemberTraffic // contracts. While the top-up trigger for SV validators is disabled by default, we also // explicitly ignore SV related MemberTraffic contracts here as a safeguard for the case of // 3rd party top-ups of SV nodes or an SV validator misconfiguration that changes the // defaults. - None + Left("it is an SV node, which is granted unlimited traffic") } else { - Some( + Right( rulesAndStates.dsoRules.payload.initialTrafficState.asScala .get(memberId.toProtoPrimitive) .fold(0L)(_.consumedTraffic) From 8fd620cbf90805c01454be96d5da2d03b2608a97 Mon Sep 17 00:00:00 2001 From: sadiq1971 Date: Wed, 12 Aug 2026 01:49:39 +0600 Subject: [PATCH 3/5] refactor(scala-sv): keep the traffic trigger skip path independent of the sequencer [ci] The one-off target check was sequenced into the skip, so an unreachable sequencer could make skipping a foreign contract retry or fail. It is now best-effort, guarded by a compare-and-set so only one check runs at a time and it logs at most once, and it reuses a single helper for resolving what the sequencer serves. Signed-off-by: sadiq1971 --- ...cerLimitWithMemberTrafficTriggerBase.scala | 109 ++++++++++-------- 1 file changed, 63 insertions(+), 46 deletions(-) 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 f1374338d2..78e26676df 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 @@ -18,6 +18,7 @@ import org.lfdecentralizedtrust.splice.util.AssignedContract import java.util.concurrent.atomic.AtomicBoolean import scala.concurrent.{ExecutionContext, Future} +import scala.util.control.NonFatal /** Grants the traffic purchased via `MemberTraffic` contracts on the sequencer of a single * synchronizer. @@ -81,8 +82,10 @@ abstract class ReconcileSequencerLimitWithMemberTrafficTriggerBase( tc: TraceContext ): Future[Either[String, Long]] - /** Guards the one-off check in [[warnOnceIfTargetNotServed]]. */ - private val targetChecked = new AtomicBoolean(false) + /** Set while a target check is in flight or has produced a conclusive answer, so that + * [[warnOnceIfTargetNotServed]] logs at most once and runs at most one check at a time. + */ + private val targetCheckSettled = new AtomicBoolean(false) override def completeTask( memberTraffic: AssignedContract[ @@ -107,75 +110,89 @@ abstract class ReconcileSequencerLimitWithMemberTrafficTriggerBase( ), synchronizerId => if (synchronizerId != targetSynchronizerId) { + val outcome = TaskSuccess( + s"Skipping MemberTraffic contract for synchronizer $synchronizerId, " + + s"this trigger reconciles $targetSynchronizerId" + ) // A misconfigured target makes every contract land here, which would otherwise - // leave the trigger silently granting nothing, so check the target itself once. - warnOnceIfTargetNotServed().map { _ => - TaskSuccess( - s"Skipping MemberTraffic contract for synchronizer $synchronizerId, " + - s"this trigger reconciles $targetSynchronizerId" - ) - } + // leave the trigger silently granting nothing. The check is best-effort so that + // skipping never depends on the sequencer being reachable. + warnOnceIfTargetNotServed().map(_ => outcome) } else { - reconcileMember(memberId) + processMember(memberId) }, ), ) } - private def reconcileMember(memberId: Member)(implicit tc: TraceContext): Future[TaskOutcome] = + private def processMember(memberId: Member)(implicit tc: TraceContext): Future[TaskOutcome] = trafficLimitOffset(memberId).flatMap { case Left(reason) => Future.successful(TaskSuccess(s"Skipping MemberTraffic contract for $memberId: $reason")) case Right(offset) => - sequencerAdminConnection().flatMap { sequencerAdminConnection => - sequencerAdminConnection.getStatus - .map(_.successOption.map(_.synchronizerId)) - .flatMap { - case None => - Future.failed( - Status.FAILED_PRECONDITION - .withDescription("Sequencer is not yet initialized") - .asRuntimeException() - ) - case Some(sequencerSynchronizerId) - if sequencerSynchronizerId.logical != targetSynchronizerId => - // Granting here would credit a sequencer of a different synchronizer. Reported as - // retryable so that a connection that is still switching over recovers on its own - // instead of dropping the purchase. - Future.failed( - Status.FAILED_PRECONDITION - .withDescription( - s"The sequencer admin connection serves ${sequencerSynchronizerId.logical}, " + - s"but this trigger reconciles $targetSynchronizerId" - ) - .asRuntimeException() + servedSynchronizerId().flatMap { + case (_, None) => + Future.failed( + Status.FAILED_PRECONDITION + .withDescription("Sequencer is not yet initialized") + .asRuntimeException() + ) + case (_, Some(served)) if served != targetSynchronizerId => + // Granting here would credit a sequencer of a different synchronizer. Reported as + // retryable so that a connection that is still switching over recovers on its own + // instead of dropping the purchase. A target that is permanently wrong therefore + // retries rather than failing once, and is reported by warnOnceIfTargetNotServed. + Future.failed( + Status.FAILED_PRECONDITION + .withDescription( + s"The sequencer admin connection serves $served, " + + s"but this trigger reconciles $targetSynchronizerId" ) - case _ => - reconcileExtraTrafficLimitForMember(memberId, offset, sequencerAdminConnection) - } + .asRuntimeException() + ) + case (sequencerAdminConnection, _) => + reconcileExtraTrafficLimitForMember(memberId, offset, sequencerAdminConnection) } } + /** The connection we would grant on, together with the synchronizer it serves. The synchronizer + * is absent while the sequencer is still initializing. + */ + private def servedSynchronizerId()(implicit + tc: TraceContext + ): Future[(SequencerAdminConnection, Option[SynchronizerId])] = + sequencerAdminConnection().flatMap { connection => + connection.getStatus.map { status => + (connection, status.successOption.map(_.synchronizerId.logical)) + } + } + /** Logs at most once if the sequencer we would grant on does not serve [[targetSynchronizerId]]. - * A sequencer that is not initialized yet leaves the check pending for a later task. + * Best-effort: an unreachable or uninitialized sequencer leaves the check for a later task and + * never fails the caller. */ private def warnOnceIfTargetNotServed()(implicit tc: TraceContext): Future[Unit] = - if (targetChecked.get()) { + if (!targetCheckSettled.compareAndSet(false, true)) { Future.unit } else { - sequencerAdminConnection() - .flatMap(_.getStatus) - .map(_.successOption.map(_.synchronizerId)) + // delegate so that a subclass throwing instead of failing its future is still caught below + Future + .delegate(servedSynchronizerId()) .map { - case Some(sequencerSynchronizerId) => - if (sequencerSynchronizerId.logical != targetSynchronizerId) { + case (_, Some(served)) => + if (served != targetSynchronizerId) { logger.warn( s"This trigger reconciles $targetSynchronizerId, but its sequencer serves " + - s"${sequencerSynchronizerId.logical}, so no traffic will ever be granted" + s"$served, so no traffic is granted while that remains the case" ) } - targetChecked.set(true) - case None => () + case (_, None) => + // Inconclusive, so allow a later task to check again. + targetCheckSettled.set(false) + } + .recover { case NonFatal(e) => + logger.debug(s"Could not check the sequencer of $targetSynchronizerId: $e") + targetCheckSettled.set(false) } } From 6a6372145076b935f1965a40ca71929e4ed6d1fb Mon Sep 17 00:00:00 2001 From: sadiq1971 Date: Thu, 20 Aug 2026 04:53:23 +0600 Subject: [PATCH 4/5] address review: warn on the miswired-connection path, note the BFT shape [ci] The mismatch branch claimed to be reported by warnOnceIfTargetNotServed but never called it, so a wrong connection produced retry noise with no diagnosis. It already knows what the sequencer serves, so it now warns directly through a shared at-most-once helper. Splits the single settled flag in two: confirming the wiring must not consume the one warning a later mismatch is entitled to. Also sketches the BFT extension on the class docstring, as #31 asks. Signed-off-by: sadiq1971 --- ...cerLimitWithMemberTrafficTriggerBase.scala | 56 ++++++++++++------- 1 file changed, 37 insertions(+), 19 deletions(-) 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 78e26676df..b6eacfce89 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 @@ -28,6 +28,12 @@ import scala.util.control.NonFatal * synchronizer, given by [[targetSynchronizerId]], and skips everything else; the operator of * another synchronizer grants those purchases on its own sequencer. * + * One instance reconciles against one sequencer, which is the MVP shape. A BFT synchronizer has + * several, and a traffic grant is aggregated across the sequencer group and only commits once its + * threshold is met, so its operator runs one instance of this trigger per sequencer, each with its + * own [[sequencerAdminConnection]]. Nothing here assumes the single-sequencer case beyond that + * hook returning one connection. + * * [[targetSynchronizerId]] is a stable value for the lifetime of the trigger, so a node that * changes the logical synchronizer it serves has to be restarted. Note that the sibling * `SvOnboardingUnlimitedTrafficTrigger` instead re-resolves the active synchronizer from @@ -82,10 +88,14 @@ abstract class ReconcileSequencerLimitWithMemberTrafficTriggerBase( tc: TraceContext ): Future[Either[String, Long]] - /** Set while a target check is in flight or has produced a conclusive answer, so that - * [[warnOnceIfTargetNotServed]] logs at most once and runs at most one check at a time. + /** Set once the mismatch has been logged, so that it is reported once rather than per task. */ + private val targetMismatchWarned = new AtomicBoolean(false) + + /** Set once the sequencer has been seen serving [[targetSynchronizerId]], so that the skip path + * stops looking it up. Deliberately separate from [[targetMismatchWarned]]: confirming the + * wiring must not consume the one warning a later mismatch is entitled to. */ - private val targetCheckSettled = new AtomicBoolean(false) + private val targetConfirmed = new AtomicBoolean(false) override def completeTask( memberTraffic: AssignedContract[ @@ -140,8 +150,9 @@ abstract class ReconcileSequencerLimitWithMemberTrafficTriggerBase( case (_, Some(served)) if served != targetSynchronizerId => // Granting here would credit a sequencer of a different synchronizer. Reported as // retryable so that a connection that is still switching over recovers on its own - // instead of dropping the purchase. A target that is permanently wrong therefore - // retries rather than failing once, and is reported by warnOnceIfTargetNotServed. + // instead of dropping the purchase. A permanent miswiring therefore retries rather + // than failing once, so warn alongside it to give the retries a one-line diagnosis. + warnOnceTargetNotServed(served) Future.failed( Status.FAILED_PRECONDITION .withDescription( @@ -167,32 +178,39 @@ abstract class ReconcileSequencerLimitWithMemberTrafficTriggerBase( } } - /** Logs at most once if the sequencer we would grant on does not serve [[targetSynchronizerId]]. - * Best-effort: an unreachable or uninitialized sequencer leaves the check for a later task and - * never fails the caller. + /** Logs at most once that the sequencer we would grant on serves `served` rather than + * [[targetSynchronizerId]]. + */ + private def warnOnceTargetNotServed(served: SynchronizerId)(implicit tc: TraceContext): Unit = + if (targetMismatchWarned.compareAndSet(false, true)) { + logger.warn( + s"This trigger reconciles $targetSynchronizerId, but its sequencer serves " + + s"$served, so no traffic is granted while that remains the case" + ) + } + + /** Looks the sequencer up to warn on a mismatch, for callers that do not already know what it + * serves. Best-effort: an unreachable or uninitialized sequencer leaves the check for a later + * task and never fails the caller. */ private def warnOnceIfTargetNotServed()(implicit tc: TraceContext): Future[Unit] = - if (!targetCheckSettled.compareAndSet(false, true)) { + if (targetConfirmed.get() || targetMismatchWarned.get()) { Future.unit } else { // delegate so that a subclass throwing instead of failing its future is still caught below Future .delegate(servedSynchronizerId()) .map { - case (_, Some(served)) => - if (served != targetSynchronizerId) { - logger.warn( - s"This trigger reconciles $targetSynchronizerId, but its sequencer serves " + - s"$served, so no traffic is granted while that remains the case" - ) - } + case (_, Some(served)) if served != targetSynchronizerId => + warnOnceTargetNotServed(served) + case (_, Some(_)) => + targetConfirmed.set(true) case (_, None) => - // Inconclusive, so allow a later task to check again. - targetCheckSettled.set(false) + // Inconclusive, so leave the check for a later task. + () } .recover { case NonFatal(e) => logger.debug(s"Could not check the sequencer of $targetSynchronizerId: $e") - targetCheckSettled.set(false) } } From 30e42abef89beabf6a0f34a1af7274d25db023f6 Mon Sep 17 00:00:00 2001 From: sadiq1971 Date: Tue, 25 Aug 2026 01:25:33 +0600 Subject: [PATCH 5/5] address review: take the synchronizer id from the sequencer [ci] Drops targetSynchronizerId and everything that existed to police it: both warn helpers, both flags and the target-vs-served mismatch branch. The foreign-contract skip now compares the contract against what the sequencer reports it serves, so there is no configured value left to disagree with. This restores the pre-extraction behaviour, where the synchronizer id was read from the connection per contract. The SV subclass keeps its own id for its store queries. Also restores the unparseable-synchronizer-id warning from #14, which a merge commit carried and the rebase onto the merged base dropped. Signed-off-by: sadiq1971 --- ...cerLimitWithMemberTrafficTriggerBase.scala | 172 ++++-------------- ...quencerLimitWithMemberTrafficTrigger.scala | 8 +- 2 files changed, 43 insertions(+), 137 deletions(-) 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 b6eacfce89..f35e514220 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 @@ -16,32 +16,11 @@ import org.lfdecentralizedtrust.splice.environment.TopologyAdminConnection.Topol import org.lfdecentralizedtrust.splice.store.AppStore import org.lfdecentralizedtrust.splice.util.AssignedContract -import java.util.concurrent.atomic.AtomicBoolean import scala.concurrent.{ExecutionContext, Future} -import scala.util.control.NonFatal -/** Grants the traffic purchased via `MemberTraffic` contracts on the sequencer of a single - * synchronizer. - * - * Traffic can be purchased for any synchronizer, so a node observes `MemberTraffic` contracts - * naming synchronizers it does not serve. Each instance of this trigger owns exactly one - * synchronizer, given by [[targetSynchronizerId]], and skips everything else; the operator of - * another synchronizer grants those purchases on its own sequencer. - * - * One instance reconciles against one sequencer, which is the MVP shape. A BFT synchronizer has - * several, and a traffic grant is aggregated across the sequencer group and only commits once its - * threshold is met, so its operator runs one instance of this trigger per sequencer, each with its - * own [[sequencerAdminConnection]]. Nothing here assumes the single-sequencer case beyond that - * hook returning one connection. - * - * [[targetSynchronizerId]] is a stable value for the lifetime of the trigger, so a node that - * changes the logical synchronizer it serves has to be restarted. Note that the sibling - * `SvOnboardingUnlimitedTrafficTrigger` instead re-resolves the active synchronizer from - * `AmuletConfigSchedule` on every run, so the two source the same concept differently. - * - * Purchases are matched on the synchronizer id alone. Whether a purchase recorded against a - * different migration id of the same synchronizer is visible at all is decided upstream, by the - * ingestion filter of the store backing this trigger. +/** 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. * * This trigger currently relies on enough nodes working on the same set traffic balance request * around the same time. It also depends on the sorting of tasks done in OnAssignedContractTrigger @@ -65,38 +44,23 @@ abstract class ReconcileSequencerLimitWithMemberTrafficTriggerBase( splice.decentralizedsynchronizer.MemberTraffic.COMPANION, ) { - /** The synchronizer this trigger reconciles. `MemberTraffic` contracts for any other - * synchronizer are skipped. - */ - protected def targetSynchronizerId: SynchronizerId - - /** Admin connection to the sequencer serving [[targetSynchronizerId]]. */ + /** Admin connection to the sequencer this trigger grants on. */ protected def sequencerAdminConnection()(implicit tc: TraceContext ): Future[SequencerAdminConnection] - /** Total traffic purchased for `memberId` on [[targetSynchronizerId]]. */ + /** Total traffic purchased for `memberId` on the synchronizer this trigger grants on. */ protected def getTotalPurchasedMemberTraffic(memberId: Member)(implicit tc: TraceContext ): Future[Long] - /** The traffic already consumed by `memberId` before this trigger took over its reconciliation, - * added to the purchased total when setting the limit. A `Left` skips the member and carries the - * reason, which is how members that are granted unlimited traffic are excluded. + /** Traffic already consumed by `memberId` before this trigger took over, added to the purchased + * total. A `Left` skips the member and carries the reason. */ protected def trafficLimitOffset(memberId: Member)(implicit tc: TraceContext ): Future[Either[String, Long]] - /** Set once the mismatch has been logged, so that it is reported once rather than per task. */ - private val targetMismatchWarned = new AtomicBoolean(false) - - /** Set once the sequencer has been seen serving [[targetSynchronizerId]], so that the skip path - * stops looking it up. Deliberately separate from [[targetMismatchWarned]]: confirming the - * wiring must not consume the one warning a later mismatch is entitled to. - */ - private val targetConfirmed = new AtomicBoolean(false) - override def completeTask( memberTraffic: AssignedContract[ splice.decentralizedsynchronizer.MemberTraffic.ContractId, @@ -113,107 +77,51 @@ abstract class ReconcileSequencerLimitWithMemberTrafficTriggerBase( SynchronizerId .fromString(memberTraffic.payload.synchronizerId) .fold( - err => + err => { // Skip contracts with invalid synchronizer ids + logger.warn(s"Skipping MemberTraffic with unparseable synchronizerId: ${err}") Future.successful( TaskSuccess(s"Skipping MemberTraffic with invalid synchronizerId: ${err}") - ), - synchronizerId => - if (synchronizerId != targetSynchronizerId) { - val outcome = TaskSuccess( - s"Skipping MemberTraffic contract for synchronizer $synchronizerId, " + - s"this trigger reconciles $targetSynchronizerId" - ) - // A misconfigured target makes every contract land here, which would otherwise - // leave the trigger silently granting nothing. The check is best-effort so that - // skipping never depends on the sequencer being reachable. - warnOnceIfTargetNotServed().map(_ => outcome) - } else { - processMember(memberId) - }, + ) + }, + synchronizerId => processMember(memberId, synchronizerId), ), ) } - private def processMember(memberId: Member)(implicit tc: TraceContext): Future[TaskOutcome] = - trafficLimitOffset(memberId).flatMap { - case Left(reason) => - Future.successful(TaskSuccess(s"Skipping MemberTraffic contract for $memberId: $reason")) - case Right(offset) => - servedSynchronizerId().flatMap { - case (_, None) => - Future.failed( - Status.FAILED_PRECONDITION - .withDescription("Sequencer is not yet initialized") - .asRuntimeException() - ) - case (_, Some(served)) if served != targetSynchronizerId => - // Granting here would credit a sequencer of a different synchronizer. Reported as - // retryable so that a connection that is still switching over recovers on its own - // instead of dropping the purchase. A permanent miswiring therefore retries rather - // than failing once, so warn alongside it to give the retries a one-line diagnosis. - warnOnceTargetNotServed(served) - Future.failed( - Status.FAILED_PRECONDITION - .withDescription( - s"The sequencer admin connection serves $served, " + - s"but this trigger reconciles $targetSynchronizerId" - ) - .asRuntimeException() - ) - case (sequencerAdminConnection, _) => - reconcileExtraTrafficLimitForMember(memberId, offset, sequencerAdminConnection) - } - } - - /** The connection we would grant on, together with the synchronizer it serves. The synchronizer - * is absent while the sequencer is still initializing. - */ - private def servedSynchronizerId()(implicit + private def processMember(memberId: Member, synchronizerId: SynchronizerId)(implicit tc: TraceContext - ): Future[(SequencerAdminConnection, Option[SynchronizerId])] = + ): Future[TaskOutcome] = sequencerAdminConnection().flatMap { connection => - connection.getStatus.map { status => - (connection, status.successOption.map(_.synchronizerId.logical)) + connection.getStatus.map(_.successOption.map(_.synchronizerId.logical)).flatMap { + case None => + Future.failed( + Status.FAILED_PRECONDITION + .withDescription("Sequencer is not yet initialized") + .asRuntimeException() + ) + case Some(served) if served != synchronizerId => + // Traffic can be purchased for any registered synchronizer, so we observe contracts this + // sequencer does not serve. They are granted by the operator of the synchronizer they + // name, on that synchronizer's own sequencer. + Future.successful( + TaskSuccess( + s"Skipping MemberTraffic contract for synchronizer $synchronizerId, " + + s"this sequencer serves $served" + ) + ) + case Some(_) => + trafficLimitOffset(memberId).flatMap { + case Left(reason) => + Future.successful( + TaskSuccess(s"Skipping MemberTraffic contract for $memberId: $reason") + ) + case Right(offset) => + reconcileExtraTrafficLimitForMember(memberId, offset, connection) + } } } - /** Logs at most once that the sequencer we would grant on serves `served` rather than - * [[targetSynchronizerId]]. - */ - private def warnOnceTargetNotServed(served: SynchronizerId)(implicit tc: TraceContext): Unit = - if (targetMismatchWarned.compareAndSet(false, true)) { - logger.warn( - s"This trigger reconciles $targetSynchronizerId, but its sequencer serves " + - s"$served, so no traffic is granted while that remains the case" - ) - } - - /** Looks the sequencer up to warn on a mismatch, for callers that do not already know what it - * serves. Best-effort: an unreachable or uninitialized sequencer leaves the check for a later - * task and never fails the caller. - */ - private def warnOnceIfTargetNotServed()(implicit tc: TraceContext): Future[Unit] = - if (targetConfirmed.get() || targetMismatchWarned.get()) { - Future.unit - } else { - // delegate so that a subclass throwing instead of failing its future is still caught below - Future - .delegate(servedSynchronizerId()) - .map { - case (_, Some(served)) if served != targetSynchronizerId => - warnOnceTargetNotServed(served) - case (_, Some(_)) => - targetConfirmed.set(true) - case (_, None) => - // Inconclusive, so leave the check for a later task. - () - } - .recover { case NonFatal(e) => - logger.debug(s"Could not check the sequencer of $targetSynchronizerId: $e") - } - } - private def reconcileExtraTrafficLimitForMember( memberId: Member, offset: Long, diff --git a/apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/automation/singlesv/ReconcileSequencerLimitWithMemberTrafficTrigger.scala b/apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/automation/singlesv/ReconcileSequencerLimitWithMemberTrafficTrigger.scala index 87cf3728ec..63193cdf01 100644 --- a/apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/automation/singlesv/ReconcileSequencerLimitWithMemberTrafficTrigger.scala +++ b/apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/automation/singlesv/ReconcileSequencerLimitWithMemberTrafficTrigger.scala @@ -29,7 +29,7 @@ class ReconcileSequencerLimitWithMemberTrafficTrigger( override protected val context: TriggerContext, store: SvDsoStore, synchronizerNodeService: SynchronizerNodeService[LocalSynchronizerNode], - override protected val targetSynchronizerId: SynchronizerId, + synchronizerId: SynchronizerId, trafficBalanceReconciliationDelay: NonNegativeFiniteDuration, )(implicit ec: ExecutionContext, @@ -48,15 +48,13 @@ class ReconcileSequencerLimitWithMemberTrafficTrigger( override protected def getTotalPurchasedMemberTraffic(memberId: Member)(implicit tc: TraceContext ): Future[Long] = - store.getTotalPurchasedMemberTraffic(memberId, targetSynchronizerId) + store.getTotalPurchasedMemberTraffic(memberId, synchronizerId) override protected def trafficLimitOffset(memberId: Member)(implicit tc: TraceContext ): Future[Either[String, Long]] = store.getDsoRulesWithSvNodeStates().map { rulesAndStates => - if ( - rulesAndStates.activeSvParticipantAndMediatorIds(targetSynchronizerId).contains(memberId) - ) { + if (rulesAndStates.activeSvParticipantAndMediatorIds(synchronizerId).contains(memberId)) { // SVs are granted unlimited traffic and do not need to purchase it via MemberTraffic // contracts. While the top-up trigger for SV validators is disabled by default, we also // explicitly ignore SV related MemberTraffic contracts here as a safeguard for the case of