Merge pull request #11280 from vegaprotocol/11279

fix: handle properly multiple transfers (potentially recurring vs gov…

Author: ze97286

CHANGELOG.md

@@ -18,6 +18,7 @@
 ### 🐛 Fixes
 
 - [11066](https://github.com/vegaprotocol/vega/issues/11066) - Ensure vesting statistics match vesting accounts numbers.
+- [11279](https://github.com/vegaprotocol/vega/issues/11279) - Handle properly the case of multiple transfers for the same game id. 
 
 
 ## 0.76.1

core/banking/engine.go

@@ -210,6 +210,9 @@ type Engine struct {
 
 	metricUpdateFrequency time.Duration
 	nextMetricUpdate      time.Time
+
+	// transient cache used to market a dispatch strategy as checked for eligibility for this round so we don't check again.
+	dispatchRequiredCache map[string]bool
 }
 
 type withdrawalRef struct {
@@ -275,6 +278,7 @@ func New(log *logging.Logger,
 		pendingPerAssetAndPartyFeeDiscountUpdates: map[string]map[string]*num.Uint{},
 		primaryBridgeView:                         primaryBridgeView,
 		secondaryBridgeView:                       secondaryBridgeView,
+		dispatchRequiredCache:                     map[string]bool{},
 	}
 }
 
@@ -347,6 +351,7 @@ func (e *Engine) OnEpoch(ctx context.Context, ep types.Epoch) {
 		e.distributeRecurringGovernanceTransfers(ctx)
 		e.applyPendingFeeDiscountsUpdates(ctx)
 		e.sendTeamsStats(ctx, ep.Seq)
+		e.dispatchRequiredCache = map[string]bool{}
 		// as the metrics are going to be published here, we want to progress the next update.
 		e.nextMetricUpdate = e.timeService.GetTimeNow().Add(e.metricUpdateFrequency)
 	default:

core/banking/recurring_transfers.go

@@ -185,6 +185,11 @@ func (e *Engine) ensureNoRecurringTransferDuplicates(
 // NB2: for validator ranking this will always return true as it is assumed that for the network to resume there must always be
 // a validator with non zero ranking.
 func (e *Engine) dispatchRequired(ctx context.Context, ds *vegapb.DispatchStrategy) bool {
+	required, ok := e.dispatchRequiredCache[e.hashDispatchStrategy(ds)]
+	if ok {
+		return required
+	}
+	defer func() { e.dispatchRequiredCache[e.hashDispatchStrategy(ds)] = required }()
 	switch ds.Metric {
 	case vegapb.DispatchMetric_DISPATCH_METRIC_MAKER_FEES_PAID,
 		vegapb.DispatchMetric_DISPATCH_METRIC_MAKER_FEES_RECEIVED,
@@ -210,17 +215,21 @@ func (e *Engine) dispatchRequired(ctx context.Context, ds *vegapb.DispatchStrate
 					break
 				}
 			}
-			return hasNonZeroMetric || (hasEligibleParties && ds.DistributionStrategy == vegapb.DistributionStrategy_DISTRIBUTION_STRATEGY_RANK)
+			required = hasNonZeroMetric || (hasEligibleParties && ds.DistributionStrategy == vegapb.DistributionStrategy_DISTRIBUTION_STRATEGY_RANK)
+			return required
 		} else {
 			tcs, pcs := e.marketActivityTracker.CalculateMetricForTeams(ctx, ds)
 			gs := events.NewTeamGameScoresEvent(ctx, int64(e.currentEpoch), e.hashDispatchStrategy(ds), e.timeService.GetTimeNow(), tcs, pcs)
 			e.broker.Send(gs)
-			return len(tcs) > 0
+			required = len(tcs) > 0
+			return required
 		}
 	case vegapb.DispatchMetric_DISPATCH_METRIC_VALIDATOR_RANKING:
-		return true
+		required = true
+		return required
 	}
-	return false
+	required = false
+	return required
 }
 
 func (e *Engine) distributeRecurringTransfers(ctx context.Context, newEpoch uint64) {