Fix minor issues with CCTPv2 Observer (#1448)

- Remove a metrics tag that has unbounded cardinality
- Don't keep calculating deposit hashes if ctx is cancelled

Author: rstout

execute/tokendata/cctp/v2/http_client_test.go

@@ -57,8 +57,8 @@ func (m *MockMetricsReporter) TrackAttestationAPILatency(
 	m.Called(sourceChain, sourceDomain, success, httpStatus, latency)
 }
 
-func (m *MockMetricsReporter) TrackObserveLatency(numRequests int, latency time.Duration) {
-	m.Called(numRequests, latency)
+func (m *MockMetricsReporter) TrackObserveLatency(latency time.Duration) {
+	m.Called(latency)
 }
 
 func (m *MockMetricsReporter) TrackDepositHashCalculationError(

execute/tokendata/cctp/v2/metrics.go

@@ -41,7 +41,7 @@ var (
 			Help:    "Latency of CCTP v2 Observe() calls",
 			Buckets: prometheus.DefBuckets,
 		},
-		[]string{"destChainFamily", "destChainID", "numRequests"},
+		[]string{"destChainFamily", "destChainID"},
 	)
 
 	PromCCTPv2DepositHashCalculationErrorCounter = promauto.NewCounterVec(
@@ -73,7 +73,7 @@ var (
 type MetricsReporter interface {
 	TrackAttestationAPILatency(
 		sourceChain cciptypes.ChainSelector, sourceDomain uint32, success bool, httpStatus string, latency time.Duration)
-	TrackObserveLatency(numRequests int, latency time.Duration)
+	TrackObserveLatency(latency time.Duration)
 	TrackDepositHashCalculationError(sourceChain cciptypes.ChainSelector, sourceDomain uint32)
 	TrackMessageToTokenDataError(sourceChain cciptypes.ChainSelector, sourceDomain uint32)
 	TrackAssignTokenDataFailure(sourceChain cciptypes.ChainSelector, sourceDomain uint32)
@@ -92,7 +92,7 @@ type noOpMetricsReporter struct{}
 func (n *noOpMetricsReporter) TrackAttestationAPILatency(cciptypes.ChainSelector, uint32, bool, string, time.Duration) {
 }
 
-func (n *noOpMetricsReporter) TrackObserveLatency(int, time.Duration) {
+func (n *noOpMetricsReporter) TrackObserveLatency(time.Duration) {
 }
 
 func (n *noOpMetricsReporter) TrackDepositHashCalculationError(cciptypes.ChainSelector, uint32) {
@@ -136,9 +136,9 @@ func (r *MetricsReporterImpl) TrackAttestationAPILatency(
 }
 
 // TrackObserveLatency tracks the overall latency of Observe() calls
-func (r *MetricsReporterImpl) TrackObserveLatency(numRequests int, latency time.Duration) {
+func (r *MetricsReporterImpl) TrackObserveLatency(latency time.Duration) {
 	PromCCTPv2ObserveLatencyHistogram.
-		WithLabelValues(r.destChainFamily, r.destChainID, strconv.Itoa(numRequests)).
+		WithLabelValues(r.destChainFamily, r.destChainID).
 		Observe(latency.Seconds())
 }
 

execute/tokendata/cctp/v2/observer.go

@@ -115,7 +115,7 @@ func (o *CCTPv2TokenDataObserver) Observe(
 	startTime := time.Now()
 	defer func() {
 		latency := time.Since(startTime)
-		o.metricsReporter.TrackObserveLatency(cctpV2RequestParams.Cardinality(), latency)
+		o.metricsReporter.TrackObserveLatency(latency)
 	}()
 
 	// Extract the CCTPv2 API requests that need to be made
@@ -265,6 +265,11 @@ func (o *CCTPv2TokenDataObserver) convertCCTPv2MessagesToTokenData(
 
 		// Process each individual CCTP v2 message in the batch
 		for _, msg := range messages.Messages {
+			// Return immediately if context is cancelled to avoid heavy processing done by
+			// calculateDepositHashFn
+			if ctx.Err() != nil {
+				return result
+			}
 			// Calculate the deposit hash for this message
 			depositHash, err := o.calculateDepositHashFn(msg.DecodedMessage)
 			if err != nil {