refactor: adding locks and moving checks

core/data.go

@@ -599,8 +599,6 @@ func ConvertToPaymentMetadata(ph *commonpb.PaymentHeader) *PaymentMetadata {
 	}
 }
 
-// OperatorInfo contains information about an operator which is stored on the blockchain state,
-// corresponding to a particular quorum
 type ActiveReservation = paymentvault.IPaymentVaultReservation
 
 type OnDemandPayment struct {

core/eth/reader.go

@@ -673,14 +673,10 @@ func (t *Reader) GetActiveReservations(ctx context.Context, accountIDs []gethcom
 
 	// since reservations are returned in the same order as the accountIDs, we can directly map them
 	for i, reservation := range reservations {
-		reservationsMap[accountIDs[i]] = &reservation
-	}
-
-	// filter out all zero-valued reservations
-	for accountID, reservation := range reservationsMap {
-		if isZeroValuedReservation(*reservation) {
-			delete(reservationsMap, accountID)
+		if isZeroValuedReservation(reservation) {
+			delete(reservationsMap, accountIDs[i])
 		}
+		reservationsMap[accountIDs[i]] = &reservation
 	}
 
 	return reservationsMap, nil
@@ -716,18 +712,14 @@ func (t *Reader) GetOnDemandPayments(ctx context.Context, accountIDs []gethcommo
 
 	// since payments are returned in the same order as the accountIDs, we can directly map them
 	for i, payment := range payments {
+		if payment.Cmp(big.NewInt(0)) == 0 {
+			delete(paymentsMap, accountIDs[i])
+		}
 		paymentsMap[accountIDs[i]] = &core.OnDemandPayment{
 			CumulativePayment: payment,
 		}
 	}
 
-	// filter out all zero-valued payments
-	for accountID, payment := range paymentsMap {
-		if payment.CumulativePayment.Cmp(big.NewInt(0)) == 0 {
-			delete(paymentsMap, accountID)
-		}
-	}
-
 	return paymentsMap, nil
 }
 
@@ -741,7 +733,7 @@ func (t *Reader) GetOnDemandPaymentByAccount(ctx context.Context, accountID geth
 	if err != nil {
 		return nil, err
 	}
-	if onDemandPayment == big.NewInt(0) {
+	if onDemandPayment.Cmp(big.NewInt(0)) == 0 {
 		return nil, errors.New("ondemand payment does not exist for given account")
 	}
 	return &core.OnDemandPayment{

core/meterer/onchain_state.go

@@ -146,6 +146,8 @@ func (pcs *OnchainPaymentState) RefreshOnchainPaymentState(ctx context.Context,
 
 // GetActiveReservationByAccount returns a pointer to the active reservation for the given account ID; no writes will be made to the reservation
 func (pcs *OnchainPaymentState) GetActiveReservationByAccount(ctx context.Context, accountID gethcommon.Address) (*core.ActiveReservation, error) {
+	pcs.ReservationsLock.RLock()
+	defer pcs.ReservationsLock.RUnlock()
 	if reservation, ok := (pcs.ActiveReservations)[accountID]; ok {
 		return reservation, nil
 	}
@@ -172,6 +174,8 @@ func (pcs *OnchainPaymentState) GetActiveReservationByAccountOnChain(ctx context
 
 // GetOnDemandPaymentByAccount returns a pointer to the on-demand payment for the given account ID; no writes will be made to the payment
 func (pcs *OnchainPaymentState) GetOnDemandPaymentByAccount(ctx context.Context, accountID gethcommon.Address) (*core.OnDemandPayment, error) {
+	pcs.OnDemandLocks.RLock()
+	defer pcs.OnDemandLocks.RUnlock()
 	if payment, ok := (pcs.OnDemandPayments)[accountID]; ok {
 		return payment, nil
 	}