refactor: quorum specific metrics for metered bytes (#1668)

* feat: quorum specific metrics for metered bytes

* fix: unit test mock calls

Author: hopeyen

core/meterer/meterer.go

@@ -73,41 +73,35 @@ func (m *Meterer) Start(ctx context.Context) {
 
 // MeterRequest validates a blob header and adds it to the meterer's state
 // TODO: return error if there's a rejection (with reasoning) or internal error (should be very rare)
-func (m *Meterer) MeterRequest(ctx context.Context, header core.PaymentMetadata, numSymbols uint64, quorumNumbers []uint8, receivedAt time.Time) (uint64, error) {
+func (m *Meterer) MeterRequest(ctx context.Context, header core.PaymentMetadata, numSymbols uint64, quorumNumbers []uint8, receivedAt time.Time) (map[core.QuorumID]uint64, error) {
 	m.logger.Info("Validating incoming request's payment metadata", "paymentMetadata", header, "numSymbols", numSymbols, "quorumNumbers", quorumNumbers)
 
 	params, err := m.ChainPaymentState.GetPaymentGlobalParams()
 	if err != nil {
-		return 0, fmt.Errorf("failed to get payment global params: %w", err)
+		return nil, fmt.Errorf("failed to get payment global params: %w", err)
 	}
 	// Validate against the payment method
 	if !payment_logic.IsOnDemandPayment(&header) {
 		reservations, err := m.ChainPaymentState.GetReservedPaymentByAccountAndQuorums(ctx, header.AccountID, quorumNumbers)
 		if err != nil {
-			return 0, fmt.Errorf("failed to get active reservation by account: %w", err)
+			return nil, fmt.Errorf("failed to get active reservation by account: %w", err)
 		}
-		if err := m.serveReservationRequest(ctx, params, header, reservations, numSymbols, quorumNumbers, receivedAt); err != nil {
-			return 0, fmt.Errorf("invalid reservation request: %w", err)
-		}
-	} else {
-		onDemandPayment, err := m.ChainPaymentState.GetOnDemandPaymentByAccount(ctx, header.AccountID)
+		usage, err := m.serveReservationRequest(ctx, params, header, reservations, numSymbols, quorumNumbers, receivedAt)
 		if err != nil {
-			return 0, fmt.Errorf("failed to get on-demand payment by account: %w", err)
-		}
-		if err := m.serveOnDemandRequest(ctx, params, header, onDemandPayment, numSymbols, quorumNumbers, receivedAt); err != nil {
-			return 0, fmt.Errorf("invalid on-demand request: %w", err)
+			return nil, fmt.Errorf("invalid reservation request: %w", err)
 		}
+		return usage, nil
 	}
 
-	// TODO(hopeyen): each quorum can have different min num symbols; the returned symbolsCharged is only for used for metrics.
-	// for now we simply return the charge for quorum 0, as quorums are likely to share the same min num symbols
-	// we can make this more granular by adding metrics to the meterer later on
-	_, protocolConfig, err := params.GetQuorumConfigs(OnDemandQuorumID)
+	onDemandPayment, err := m.ChainPaymentState.GetOnDemandPaymentByAccount(ctx, header.AccountID)
 	if err != nil {
-		return 0, fmt.Errorf("failed to get on-demand quorum config: %w", err)
+		return nil, fmt.Errorf("failed to get on-demand payment by account: %w", err)
 	}
-	symbolsCharged := payment_logic.SymbolsCharged(numSymbols, protocolConfig.MinNumSymbols)
-	return symbolsCharged, nil
+	usage, err := m.serveOnDemandRequest(ctx, params, header, onDemandPayment, numSymbols, quorumNumbers, receivedAt)
+	if err != nil {
+		return nil, fmt.Errorf("invalid on-demand request: %w", err)
+	}
+	return usage, nil
 }
 
 // serveReservationRequest handles the rate limiting logic for incoming requests
@@ -119,17 +113,18 @@ func (m *Meterer) serveReservationRequest(
 	numSymbols uint64,
 	quorumNumbers []uint8,
 	receivedAt time.Time,
-) error {
+) (map[core.QuorumID]uint64, error) {
 	m.logger.Debug("Recording and validating reservation usage", "header", header, "reservation", reservations)
 	if err := payment_logic.ValidateReservations(reservations, globalParams.QuorumProtocolConfigs, quorumNumbers, header.Timestamp, receivedAt.UnixNano()); err != nil {
-		return fmt.Errorf("invalid reservation: %w", err)
+		return nil, fmt.Errorf("invalid reservation: %w", err)
 	}
 
 	// Make atomic batched updates over all reservations identified by the same account and quorum
-	if err := m.incrementBinUsage(ctx, header, reservations, globalParams, numSymbols); err != nil {
-		return fmt.Errorf("failed to increment bin usages: %w", err)
+	usage, err := m.incrementBinUsage(ctx, header, reservations, globalParams, numSymbols)
+	if err != nil {
+		return nil, fmt.Errorf("failed to increment bin usages: %w", err)
 	}
-	return nil
+	return usage, nil
 }
 
 // incrementBinUsage increments the bin usage atomically and checks for overflow
@@ -138,15 +133,15 @@ func (m *Meterer) incrementBinUsage(
 	reservations map[core.QuorumID]*core.ReservedPayment,
 	globalParams *PaymentVaultParams,
 	numSymbols uint64,
-) error {
+) (map[core.QuorumID]uint64, error) {
 	charges := make(map[core.QuorumID]uint64)
 	quorumNumbers := make([]core.QuorumID, 0, len(reservations))
 	reservationWindows := make(map[core.QuorumID]uint64, len(reservations))
 	requestReservationPeriods := make(map[core.QuorumID]uint64, len(reservations))
 	for quorumID := range reservations {
 		_, protocolConfig, err := globalParams.GetQuorumConfigs(quorumID)
 		if err != nil {
-			return fmt.Errorf("failed to get quorum config for quorum %d: %w", quorumID, err)
+			return nil, fmt.Errorf("failed to get quorum config for quorum %d: %w", quorumID, err)
 		}
 		charges[quorumID] = payment_logic.SymbolsCharged(numSymbols, protocolConfig.MinNumSymbols)
 		quorumNumbers = append(quorumNumbers, quorumID)
@@ -155,13 +150,9 @@ func (m *Meterer) incrementBinUsage(
 	}
 	// Batch increment all quorums for the current quorums' reservation period
 	// For each quorum, increment by its specific symbolsCharged value
-	updatedUsages := make(map[core.QuorumID]uint64)
 	usage, err := m.MeteringStore.IncrementBinUsages(ctx, header.AccountID, quorumNumbers, requestReservationPeriods, charges)
 	if err != nil {
-		return err
-	}
-	for _, quorumID := range quorumNumbers {
-		updatedUsages[quorumID] = usage[quorumID]
+		return nil, err
 	}
 	overflowAmounts := make(map[core.QuorumID]uint64)
 	overflowPeriods := make(map[core.QuorumID]uint64)
@@ -170,28 +161,28 @@ func (m *Meterer) incrementBinUsage(
 		reservationWindow := reservationWindows[quorumID]
 		requestReservationPeriod := requestReservationPeriods[quorumID]
 		usageLimit := payment_logic.GetBinLimit(reservation.SymbolsPerSecond, reservationWindow)
-		newUsage, ok := updatedUsages[quorumID]
+		newUsage, ok := usage[quorumID]
 		if !ok {
-			return fmt.Errorf("failed to get updated usage for quorum %d", quorumID)
+			return nil, fmt.Errorf("failed to get updated usage for quorum %d", quorumID)
 		}
 		prevUsage := newUsage - charges[quorumID]
 		if newUsage <= usageLimit {
 			continue
 		} else if prevUsage >= usageLimit {
 			// Bin was already filled before this increment
-			return fmt.Errorf("bin has already been filled for quorum %d", quorumID)
+			return nil, fmt.Errorf("bin has already been filled for quorum %d", quorumID)
 		}
 		overflowPeriod := payment_logic.GetOverflowPeriod(requestReservationPeriod, reservationWindow)
 		if charges[quorumID] <= usageLimit && overflowPeriod <= payment_logic.GetReservationPeriod(int64(reservation.EndTimestamp), reservationWindow) {
 			// Needs to go to overflow bin
 			overflowAmounts[quorumID] = newUsage - usageLimit
 			overflowPeriods[quorumID] = overflowPeriod
 		} else {
-			return fmt.Errorf("overflow usage exceeds bin limit for quorum %d", quorumID)
+			return nil, fmt.Errorf("overflow usage exceeds bin limit for quorum %d", quorumID)
 		}
 	}
 	if len(overflowAmounts) != len(overflowPeriods) {
-		return fmt.Errorf("overflow amount and period mismatch")
+		return nil, fmt.Errorf("overflow amount and period mismatch")
 	}
 	// Batch increment overflow bins for all overflown reservation candidates
 	if len(overflowAmounts) > 0 {
@@ -205,55 +196,60 @@ func (m *Meterer) incrementBinUsage(
 			// Rollback the increments for the current periods
 			rollbackErr := m.MeteringStore.DecrementBinUsages(ctx, header.AccountID, quorumNumbers, requestReservationPeriods, charges)
 			if rollbackErr != nil {
-				return fmt.Errorf("failed to increment overflow bins: %w; rollback also failed: %v", err, rollbackErr)
+				return nil, fmt.Errorf("failed to increment overflow bins: %w; rollback also failed: %v", err, rollbackErr)
 			}
-			return fmt.Errorf("failed to increment overflow bins: %w; successfully rolled back increments", err)
+			return nil, fmt.Errorf("failed to increment overflow bins: %w; successfully rolled back increments", err)
 		}
 	}
 
-	return nil
+	return charges, nil
 }
 
 // serveOnDemandRequest handles the rate limiting logic for incoming requests
 // On-demand requests doesn't have additional quorum settings and should only be
 // allowed by ETH and EIGEN quorums
-func (m *Meterer) serveOnDemandRequest(ctx context.Context, globalParams *PaymentVaultParams, header core.PaymentMetadata, onDemandPayment *core.OnDemandPayment, symbolsCharged uint64, headerQuorums []uint8, receivedAt time.Time) error {
+func (m *Meterer) serveOnDemandRequest(ctx context.Context, globalParams *PaymentVaultParams, header core.PaymentMetadata, onDemandPayment *core.OnDemandPayment, numSymbols uint64, headerQuorums []uint8, receivedAt time.Time) (map[core.QuorumID]uint64, error) {
 	m.logger.Debug("Recording and validating on-demand usage", "header", header, "onDemandPayment", onDemandPayment)
 
 	if err := payment_logic.ValidateQuorum(headerQuorums, globalParams.OnDemandQuorumNumbers); err != nil {
-		return fmt.Errorf("invalid quorum for On-Demand Request: %w", err)
+		return nil, fmt.Errorf("invalid quorum for On-Demand Request: %w", err)
 	}
 
 	// Verify that the claimed cumulative payment doesn't exceed the on-chain deposit
 	if header.CumulativePayment.Cmp(onDemandPayment.CumulativePayment) > 0 {
-		return fmt.Errorf("request claims a cumulative payment greater than the on-chain deposit")
+		return nil, fmt.Errorf("request claims a cumulative payment greater than the on-chain deposit")
 	}
 
 	paymentConfig, protocolConfig, err := globalParams.GetQuorumConfigs(OnDemandQuorumID)
 	if err != nil {
-		return fmt.Errorf("failed to get payment config for on-demand quorum: %w", err)
+		return nil, fmt.Errorf("failed to get payment config for on-demand quorum: %w", err)
 	}
 
-	symbolsCharged = payment_logic.SymbolsCharged(symbolsCharged, protocolConfig.MinNumSymbols)
+	symbolsCharged := payment_logic.SymbolsCharged(numSymbols, protocolConfig.MinNumSymbols)
 	paymentCharged := payment_logic.PaymentCharged(symbolsCharged, paymentConfig.OnDemandPricePerSymbol)
 	oldPayment, err := m.MeteringStore.AddOnDemandPayment(ctx, header, paymentCharged)
 	if err != nil {
-		return fmt.Errorf("failed to update cumulative payment: %w", err)
+		return nil, fmt.Errorf("failed to update cumulative payment: %w", err)
 	}
 
 	// Update bin usage atomically and check against bin capacity
-	if err := m.incrementGlobalBinUsage(ctx, globalParams, uint64(symbolsCharged), receivedAt); err != nil {
+	if err := m.incrementGlobalBinUsage(ctx, globalParams, symbolsCharged, receivedAt); err != nil {
 		// If global bin usage update fails, roll back the payment to its previous value
 		// The rollback will only happen if the current payment value still matches what we just wrote
 		// This ensures we don't accidentally roll back a newer payment that might have been processed
 		dbErr := m.MeteringStore.RollbackOnDemandPayment(ctx, header.AccountID, header.CumulativePayment, oldPayment)
 		if dbErr != nil {
-			return dbErr
+			return nil, dbErr
 		}
-		return fmt.Errorf("failed global rate limiting: %w", err)
+		return nil, fmt.Errorf("failed global rate limiting: %w", err)
 	}
 
-	return nil
+	// charges is applied to the header quorums
+	charges := make(map[core.QuorumID]uint64, len(headerQuorums))
+	for _, quorumID := range headerQuorums {
+		charges[core.QuorumID(quorumID)] = symbolsCharged
+	}
+	return charges, nil
 }
 
 // IncrementGlobalBinUsage increments the bin usage atomically and checks for overflow