fix: make disperser client backwards compatible (#1686)

* chore(apiserver): refactor GetPaymentState to highlight legacy conversion

* wip: attempt to make disperser_client backwards compatible

TODO: review the Claude generated getPaymentStateFromLegacyAPI... no idea if its OK or not.

* docs(apiserver): document convertAllQuorumsReplyToLegacy function

* fix(disperser_client): convertLegacyPaymentStateToNew returns error

* fix(disperser_client): set correct value for ReservationRateLimitWindow

* style: targetting -> targeting

* docs: remove outdated TODO comment

* docs: add docstring for getPaymentStateFromLegacyAPI

* fix: return error if PaymentGlobalParams==nil

* docs(proto): document GetPaymentStateReply fields

also GetPaymentStateForAllQuorumsReply fields

* docs(proto): document why GetPaymentStateRequest is separate type

from GetPaymentStateForAllQuorumsRequest

* style: remove outdated comment

* fix(disperser_client): onDemandEnabled value in convertLegacyPaymentStateToNew

* style: better error message in convertLegacyPaymentStateToNew

* refactor: convertLegacyPaymentStateToNew function

return early when no reservation exists, which makes the logic much simpler

* docs(proto): fix inaccurate cumulative_payment docstring

* style(disperser_client): set reservationAdvanceWindow to 0

in convertLegacyPaymentStateToNew

* fix(disp_client): convertLegacyPaymentStateToNew quorums setting logic

* docs(proto): remove wrong cumulative_payment comment

* style: remove unneeded comments

* style: add comment explaining why append is ok

* docs: move reservation_advance_window comment to proto

Author: samlaf

api/clients/v2/disperser_client.go

@@ -17,7 +17,10 @@ import (
 	"github.com/Layr-Labs/eigenda/encoding/rs"
 	"github.com/Layr-Labs/eigensdk-go/logging"
 	"github.com/docker/go-units"
+	gethcommon "github.com/ethereum/go-ethereum/common"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 )
 
 type DisperserClientConfig struct {
@@ -390,7 +393,135 @@ func (c *disperserClient) GetPaymentState(ctx context.Context) (*disperser_rpc.G
 		Signature: signature,
 		Timestamp: timestamp,
 	}
-	return c.client.GetPaymentStateForAllQuorums(ctx, request)
+	allQuorumsReply, err := c.client.GetPaymentStateForAllQuorums(ctx, request)
+	if err != nil {
+		// Check if error is "method not found" or "unimplemented"
+		if isMethodNotFoundError(err) {
+			// Fall back to old method
+			return c.getPaymentStateFromLegacyAPI(ctx, accountID, signature, timestamp)
+		}
+		return nil, err
+	}
+
+	return allQuorumsReply, nil
+}
+
+// this is true if we are targeting a disperser that hasn't upgraded to the new API yet.
+func isMethodNotFoundError(err error) bool {
+	if st, ok := status.FromError(err); ok {
+		return st.Code() == codes.Unimplemented
+	}
+	return false
+}
+
+// getPaymentStateFromLegacyAPI retrieves the payment state from the legacy GetPaymentState grpc method.
+// It is needed until we have upgraded all dispersers (testnet and mainnet) to the new API.
+// Check those endpoints for GetPaymentStateForAllQuorums using:
+// `grpcurl disperser-testnet-holesky.eigenda.xyz:443 list disperser.v2.Disperser`
+// `grpcurl disperser.eigenda.xyz:443 list disperser.v2.Disperser`
+func (c *disperserClient) getPaymentStateFromLegacyAPI(
+	ctx context.Context, accountID gethcommon.Address, signature []byte, timestamp uint64,
+) (*disperser_rpc.GetPaymentStateForAllQuorumsReply, error) {
+	oldRequest := &disperser_rpc.GetPaymentStateRequest{
+		AccountId: accountID.Hex(),
+		Signature: signature,
+		Timestamp: timestamp,
+	}
+
+	oldResult, err := c.client.GetPaymentState(ctx, oldRequest)
+	if err != nil {
+		return nil, err
+	}
+
+	return convertLegacyPaymentStateToNew(oldResult)
+}
+
+// convertLegacyPaymentStateToNew converts the old GetPaymentStateReply to the new GetPaymentStateForAllQuorumsReply format
+func convertLegacyPaymentStateToNew(legacyReply *disperser_rpc.GetPaymentStateReply) (*disperser_rpc.GetPaymentStateForAllQuorumsReply, error) {
+
+	if legacyReply.PaymentGlobalParams == nil {
+		return nil, fmt.Errorf("legacy payment state received from disperser does not contain global params")
+	}
+	// Convert PaymentGlobalParams to PaymentVaultParams
+	var paymentVaultParams *disperser_rpc.PaymentVaultParams
+	{
+		paymentVaultParams = &disperser_rpc.PaymentVaultParams{
+			QuorumPaymentConfigs:  make(map[uint32]*disperser_rpc.PaymentQuorumConfig),
+			QuorumProtocolConfigs: make(map[uint32]*disperser_rpc.PaymentQuorumProtocolConfig),
+			OnDemandQuorumNumbers: legacyReply.PaymentGlobalParams.OnDemandQuorumNumbers,
+		}
+
+		// Apply the global params to all quorums, both on-demand and reservation.
+		onDemandQuorums := legacyReply.PaymentGlobalParams.OnDemandQuorumNumbers
+		if len(onDemandQuorums) == 0 {
+			return nil, fmt.Errorf("no on-demand quorums specified in legacy PaymentGlobalParams received from disperser")
+		}
+		reservationQuorums := legacyReply.Reservation.QuorumNumbers
+		// There may be overlapping quorums but it doesn't matter since we will apply the same global params to all of them.
+		allQuorums := append(reservationQuorums, onDemandQuorums...)
+
+		for _, quorumID := range allQuorums {
+			paymentVaultParams.QuorumPaymentConfigs[quorumID] = &disperser_rpc.PaymentQuorumConfig{
+				ReservationSymbolsPerSecond: 0, // Not available in legacy format
+				OnDemandSymbolsPerSecond:    legacyReply.PaymentGlobalParams.GlobalSymbolsPerSecond,
+				OnDemandPricePerSymbol:      legacyReply.PaymentGlobalParams.PricePerSymbol,
+			}
+
+			paymentVaultParams.QuorumProtocolConfigs[quorumID] = &disperser_rpc.PaymentQuorumProtocolConfig{
+				MinNumSymbols: legacyReply.PaymentGlobalParams.MinNumSymbols,
+				// ReservationAdvanceWindow is not used offchain at the moment so it's okay to set to any value.
+				ReservationAdvanceWindow:   0,
+				ReservationRateLimitWindow: legacyReply.PaymentGlobalParams.ReservationWindow,
+				OnDemandRateLimitWindow:    0, // Not available in legacy format
+			}
+		}
+
+		for _, quorumID := range onDemandQuorums {
+			paymentVaultParams.QuorumProtocolConfigs[quorumID].OnDemandEnabled = true
+		}
+	}
+
+	// If no reservation is available, return early with only payment vault params and cumulative payment info.
+	if legacyReply.Reservation == nil {
+		return &disperser_rpc.GetPaymentStateForAllQuorumsReply{
+			PaymentVaultParams:       paymentVaultParams,
+			CumulativePayment:        legacyReply.CumulativePayment,
+			OnchainCumulativePayment: legacyReply.OnchainCumulativePayment,
+		}, nil
+	}
+
+	// Otherwise there is a reservation available, so we need to convert it to the per-quorum format.
+
+	// We first make sure that the disperser returned valid data.
+	if len(legacyReply.PeriodRecords) == 0 {
+		return nil, fmt.Errorf("legacy payment state received from disperser does not contain period records")
+	}
+	if len(legacyReply.Reservation.QuorumNumbers) == 0 {
+		return nil, fmt.Errorf("legacy payment state received from disperser does not contain reservation quorums")
+	}
+
+	reservations := make(map[uint32]*disperser_rpc.QuorumReservation)
+	periodRecords := make(map[uint32]*disperser_rpc.PeriodRecords)
+
+	// Apply the reservation to all reservationQuorums mentioned in the reservation
+	for _, quorumID := range legacyReply.Reservation.QuorumNumbers {
+		reservations[quorumID] = &disperser_rpc.QuorumReservation{
+			SymbolsPerSecond: legacyReply.Reservation.SymbolsPerSecond,
+			StartTimestamp:   legacyReply.Reservation.StartTimestamp,
+			EndTimestamp:     legacyReply.Reservation.EndTimestamp,
+		}
+		periodRecords[quorumID] = &disperser_rpc.PeriodRecords{
+			Records: legacyReply.PeriodRecords,
+		}
+	}
+
+	return &disperser_rpc.GetPaymentStateForAllQuorumsReply{
+		PaymentVaultParams:       paymentVaultParams,
+		PeriodRecords:            periodRecords,
+		Reservations:             reservations,
+		CumulativePayment:        legacyReply.CumulativePayment,
+		OnchainCumulativePayment: legacyReply.OnchainCumulativePayment,
+	}, nil
 }
 
 // GetBlobCommitment is a utility method that calculates commitment for a blob payload.

api/grpc/disperser/v2/disperser_v2.pb.go

@@ -134,6 +134,9 @@ message BlobCommitmentReply {
 }
 
 // GetPaymentStateRequest contains parameters to query the payment state of an account.
+// GetPaymentStateForAllQuorumsRequest is a separate message type even though it currently contains the same fields,
+// because we follow buf's best practices and linting rules which recommend every RPC having its own request and reply types,
+// to allow for evolution of the API without breaking changes: https://buf.build/docs/lint/rules/#rpc_no_server_streaming
 message GetPaymentStateRequest {
   // The ID of the account being queried. This account ID is an eth wallet address of the user.
   string account_id = 1;
@@ -160,12 +163,21 @@ message GetPaymentStateReply {
   // global payment vault parameters
   PaymentGlobalParams payment_global_params = 1;
   // off-chain account reservation usage records
+  // Should be empty if reservation.quorum_numbers is empty (i.e. no reservation exists for the account).
   repeated PeriodRecord period_records = 2;
   // on-chain account reservation setting
   Reservation reservation = 3;
-  // off-chain on-demand payment usage
+  // off-chain on-demand payment usage.
+  // The bytes are parsed to a big.Int value.
+  // This value should always be <= onchain_cumulative_payment.
+  // See [common.v2.PaymentHeader.cumulative_payment] for more details.
+  //
+  // This value should only be nonzero for the EigenLabs disperser, as it is the only disperser that supports on-demand payments currently.
+  // Future work will support decentralized on-demand dispersals.
   bytes cumulative_payment = 4;
   // on-chain on-demand payment deposited
+  // The bytes are parsed to a big.Int value.
+  // See [common.v2.PaymentHeader.cumulative_payment] for more details.
   bytes onchain_cumulative_payment = 5;
 }
 
@@ -178,14 +190,23 @@ message GetPaymentStateForAllQuorumsReply {
   // payment vault parameters with per-quorum configurations
   PaymentVaultParams payment_vault_params = 1;
   // period_records maps quorum IDs to the off-chain account reservation usage records for the current and next two periods
+  // Should contain the same number of entries as the `reservations` field.
   map<uint32, PeriodRecords> period_records = 2;
   // reservations maps quorum IDs to the on-chain account reservation record
+  // Should contain the same number of entries as the `period_records` field.
   map<uint32, QuorumReservation> reservations = 3;
-  // off-chain on-demand payment usage. This field is currently only tracked by EigenLabs disperser because on-demand requests are only
-  // supported by EigenLabs. Future work will support decentralized on-demand dispersals and this field later be tracked and shared by
-  // dispersers unlimited to EigenLabs.
+  // off-chain on-demand payment usage.
+  // The bytes are parsed to a big.Int value.
+  // This value should always be <= onchain_cumulative_payment.
+  // See [common.v2.PaymentHeader.cumulative_payment] for more details.
+  //
+  // This value should only be nonzero for the EigenLabs disperser, as it is the only disperser that supports on-demand payments currently.
+  // Future work will support decentralized on-demand dispersals.
   bytes cumulative_payment = 4;
   // on-chain on-demand payment deposited.
+  // The bytes are parsed to a big.Int value.
+  // This value should always be >= cumulative_payment.
+  // See [common.v2.PaymentHeader.cumulative_payment] for more details.
   bytes onchain_cumulative_payment = 5;
 }
 
@@ -341,6 +362,8 @@ message PaymentQuorumProtocolConfig {
   uint64 min_num_symbols = 1;
 
   // reservation_advance_window is the window in seconds before a reservation starts that it can be activated
+  // It is added here for offchain to have access to all onchain data structs, but it isn't currently used,
+  // and might get removed in the future.
   uint64 reservation_advance_window = 2;
 
   // reservation_rate_limit_window is the time window in seconds for reservation rate limiting

api/proto/disperser/v2/disperser_v2.proto

@@ -290,6 +290,22 @@ func (s *DispersalServerV2) GetPaymentState(ctx context.Context, req *pb.GetPaym
 		return nil, err
 	}
 
+	return convertAllQuorumsReplyToLegacy(allQuorumsReply), nil
+}
+
+// convertAllQuorumsReplyToLegacy converts the new per-quorum payment state format to the legacy aggregated format.
+// This enables backwards compatibility by flattening multi-quorum data into a single legacy response,
+// allowing old clients to continue working properly.
+//
+// Conversion logic:
+// - PaymentVaultParams: Uses quorum 0 configuration as the global parameters (arbitrary choice)
+// - Reservations: Finds the most restrictive reservation across all quorums (minimum symbols/sec, latest start, earliest end)
+// - PeriodRecords: Selects the highest usage for each period index across all quorums
+// - OnDemand cumulative payment amounts: Passed through unchanged as they represent account-level totals
+//
+// This conversion may result in information loss for clients that need per-quorum details,
+// but preserves the most restrictive constraints to ensure client behavior remains within reservation or ondemand.
+func convertAllQuorumsReplyToLegacy(allQuorumsReply *pb.GetPaymentStateForAllQuorumsReply) *pb.GetPaymentStateReply {
 	// For PaymentVaultParams, use quorum 0 for protocol level parameters and on-demand quorum numbers
 	var paymentGlobalParams *pb.PaymentGlobalParams
 	if allQuorumsReply.PaymentVaultParams != nil &&
@@ -376,7 +392,7 @@ func (s *DispersalServerV2) GetPaymentState(ctx context.Context, req *pb.GetPaym
 		Reservation:              reservation,
 		CumulativePayment:        allQuorumsReply.CumulativePayment,
 		OnchainCumulativePayment: allQuorumsReply.OnchainCumulativePayment,
-	}, nil
+	}
 }
 
 // GetPaymentStateForAllQuorums returns payment state for all quorums including vault parameters,