Skip to content

fix: skip unparseable and foreign sync ids in traffic triggers (Scala) #14

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
sadiq1971 merged 2 commits into from
Aug 24, 2026
Merged

fix: skip unparseable and foreign sync ids in traffic triggers (Scala) #14

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
2794040
fix(scala-sv): skip unparseable and foreign sync ids in traffic trigg…
sadiq1971 Aug 10, 2026
36e365c
fix(scala-sv): warn when a sync id is unparseable [ci]
sadiq1971 Aug 13, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
52 changes: 32 additions & 20 deletions ...tralizedtrust/splice/sv/automation/delegatebased/MergeMemberTrafficContractsTrigger.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,28 +49,40 @@ class MergeMemberTrafficContractsTrigger(
// Skip contracts with invalid member ids
Future.successful(TaskSuccess(s"Skipping MemberTraffic with invalid memberId: ${err}"))
},
memberId => {
for {
dsoRules <- store.getDsoRules()
threshold = dsoRules.payload.config.numMemberTrafficContractsThreshold
synchronizerId = SynchronizerId.tryFromString(memberTraffic.payload.synchronizerId)
memberTraffics <- store.listMemberTrafficContracts(
memberId,
synchronizerId,
PageLimit.tryCreate(2 * threshold.toInt),
)
outcome <-
if (memberTraffics.length > threshold)
mergeMemberTrafficContracts(memberId, memberTraffics, controller)
else
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}")

@moritzkiefer-da moritzkiefer-da Aug 21, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can we just avoid parsing the synchronizer id in the first place? I don't really see why we shouldn't merge for an unmergeable sync id.

Future.successful(
TaskSuccess(
s"More than ${threshold} member traffic contracts are required for $memberId on domain $synchronizerId " +
s"in order to merge them. Currently, there are only ${memberTraffics.length}."
)
TaskSuccess(s"Skipping MemberTraffic with invalid synchronizerId: ${err}")
Comment thread
sadiq1971 marked this conversation as resolved.
)
} yield outcome
},
},
synchronizerId =>
for {
dsoRules <- store.getDsoRules()
threshold = dsoRules.payload.config.numMemberTrafficContractsThreshold
memberTraffics <- store.listMemberTrafficContracts(
memberId,
synchronizerId,
PageLimit.tryCreate(2 * threshold.toInt),
)
outcome <-
if (memberTraffics.length > threshold)
mergeMemberTrafficContracts(memberId, memberTraffics, controller)
else
Future.successful(
TaskSuccess(
s"More than ${threshold} member traffic contracts are required for " +
s"$memberId on domain $synchronizerId in order to merge them. " +
s"Currently, there are only ${memberTraffics.length}."
)
)
} yield outcome,
),
)
}

Expand Down
125 changes: 73 additions & 52 deletions ...trust/splice/sv/automation/singlesv/ReconcileSequencerLimitWithMemberTrafficTrigger.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -64,60 +64,81 @@ class ReconcileSequencerLimitWithMemberTrafficTrigger(
// Skip contracts with invalid member ids
Future.successful(TaskSuccess(s"Skipping MemberTraffic with invalid memberId: ${err}"))
},
memberId => {
val synchronizerId = SynchronizerId.tryFromString(memberTraffic.payload.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 =>
Future.failed(
Status.INTERNAL
.withDescription(
s"The MemberTraffic contract synchronizerId must match the connected domain ${sequencerSynchronizerId}"
)
.asRuntimeException()
)
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")
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}")
Comment thread
sadiq1971 marked this conversation as resolved.
)
},
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()
)
} else {
val trafficLimitOffset =
rulesAndStates.dsoRules.payload.initialTrafficState.asScala
.get(memberId.toProtoPrimitive)
.fold(0L)(_.consumedTraffic)
reconcileExtraTrafficLimitForMember(
memberId,
synchronizerId,
trafficLimitOffset,
sequencerAdminConnection,
)
case Some(sequencerSynchronizerId)
if sequencerSynchronizerId.logical != synchronizerId =>
// Traffic can be purchased for any registered synchronizer, so we
Comment thread
sadiq1971 marked this conversation as resolved.
// 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 " +
Comment thread
sadiq1971 marked this conversation as resolved.
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,
)
}
})
}
})
}
}
},
},
),
)
}

Expand Down
Loading