From 2794040756e472a0094de1eef2b1a3b3c66f9622 Mon Sep 17 00:00:00 2001 From: sadiq1971 Date: Mon, 10 Aug 2026 23:05:13 +0600 Subject: [PATCH 1/2] fix(scala-sv): skip unparseable and foreign sync ids in traffic triggers [ci] Traffic can be purchased for any registered synchronizer, so an SV sequencer observes MemberTraffic contracts for synchronizers it does not serve. Parse failures and foreign synchronizer ids now skip the contract instead of failing the trigger. Signed-off-by: sadiq1971 --- .../MergeMemberTrafficContractsTrigger.scala | 49 ++++--- ...quencerLimitWithMemberTrafficTrigger.scala | 122 ++++++++++-------- 2 files changed, 99 insertions(+), 72 deletions(-) diff --git a/apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/automation/delegatebased/MergeMemberTrafficContractsTrigger.scala b/apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/automation/delegatebased/MergeMemberTrafficContractsTrigger.scala index 9198f0764e..f4763fb499 100644 --- a/apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/automation/delegatebased/MergeMemberTrafficContractsTrigger.scala +++ b/apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/automation/delegatebased/MergeMemberTrafficContractsTrigger.scala @@ -49,28 +49,37 @@ 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 => + // Skip contracts with invalid synchronizer ids 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}") + ), + synchronizerId => + for { + dsoRules <- store.getDsoRules() + threshold = dsoRules.payload.config.numMemberTrafficContractsThreshold + memberTraffics <- store.listMemberTrafficContracts( + memberId, + synchronizerId, + PageLimit.tryCreate(2 * threshold.toInt), ) - ) - } yield outcome - }, + 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, + ), ) } 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 354b18cbb0..a5eb6cfb99 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 @@ -64,60 +64,78 @@ 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 => + // Skip contracts with invalid synchronizer ids + 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() ) - } 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 + // 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, + ) + } + }) } - }) - } - } - }, + }, + ), ) } From 36e365c13db934091303942faad9747e673bc304 Mon Sep 17 00:00:00 2001 From: sadiq1971 Date: Fri, 14 Aug 2026 01:04:20 +0600 Subject: [PATCH 2/2] fix(scala-sv): warn when a sync id is unparseable [ci] A foreign synchronizer id is routine once dedicated synchronizers exist, but an unparseable one means corrupt data, so it is logged as a warning as well as skipped. Matches the acceptance criterion in canton-extending-mainnet#33. Signed-off-by: sadiq1971 --- .../MergeMemberTrafficContractsTrigger.scala | 9 ++++++--- ...ReconcileSequencerLimitWithMemberTrafficTrigger.scala | 9 ++++++--- 2 files changed, 12 insertions(+), 6 deletions(-) diff --git a/apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/automation/delegatebased/MergeMemberTrafficContractsTrigger.scala b/apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/automation/delegatebased/MergeMemberTrafficContractsTrigger.scala index f4763fb499..a3bd93fcd8 100644 --- a/apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/automation/delegatebased/MergeMemberTrafficContractsTrigger.scala +++ b/apps/sv/src/main/scala/org/lfdecentralizedtrust/splice/sv/automation/delegatebased/MergeMemberTrafficContractsTrigger.scala @@ -53,11 +53,14 @@ class MergeMemberTrafficContractsTrigger( SynchronizerId .fromString(memberTraffic.payload.synchronizerId) .fold( - err => - // Skip contracts with invalid synchronizer ids + 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 => for { dsoRules <- store.getDsoRules() 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 a5eb6cfb99..90cb1a442d 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 @@ -68,11 +68,14 @@ class ReconcileSequencerLimitWithMemberTrafficTrigger( SynchronizerId .fromString(memberTraffic.payload.synchronizerId) .fold( - err => - // Skip contracts with invalid synchronizer ids + 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 =>