Skip to content

[v2] disperser client payments api #928

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 6 commits into from
Dec 6, 2024
Merged

[v2] disperser client payments api #928

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
9982bf1
feat: client construct payment header
hopeyen Dec 2, 2024
5004cf9
feat: getPaymentState and init accountant
hopeyen Dec 3, 2024
dd00e32
fix: nil checks and remove bad comment
hopeyen Dec 4, 2024
0d59ee9
fix: lint
hopeyen Dec 5, 2024
cd908e1
refactor: use protobuf getters and chain
hopeyen Dec 5, 2024
6643ca9
refactor: rm Accountant interface and rename struct to Accountant
hopeyen Dec 5, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
40 changes: 35 additions & 5 deletions api/clients/accountant.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,15 +8,15 @@ import (
"sync"
"time"

commonpb "github.com/Layr-Labs/eigenda/api/grpc/common"
disperser_rpc "github.com/Layr-Labs/eigenda/api/grpc/disperser/v2"
"github.com/Layr-Labs/eigenda/core"
"github.com/Layr-Labs/eigenda/core/meterer"
)

var requiredQuorums = []uint8{0, 1}

type Accountant interface {
AccountBlob(ctx context.Context, numSymbols uint64, quorums []uint8) (*commonpb.PaymentHeader, error)
AccountBlob(ctx context.Context, numSymbols uint64, quorums []uint8) (*core.PaymentMetadata, error)
}

var _ Accountant = &accountant{}
Expand Down Expand Up @@ -116,7 +116,7 @@ func (a *accountant) BlobPaymentInfo(ctx context.Context, numSymbols uint64, quo
}

// AccountBlob accountant provides and records payment information
func (a *accountant) AccountBlob(ctx context.Context, numSymbols uint64, quorums []uint8) (*commonpb.PaymentHeader, error) {
func (a *accountant) AccountBlob(ctx context.Context, numSymbols uint64, quorums []uint8) (*core.PaymentMetadata, error) {
binIndex, cumulativePayment, err := a.BlobPaymentInfo(ctx, numSymbols, quorums)
if err != nil {
return nil, err
Expand All @@ -127,9 +127,8 @@ func (a *accountant) AccountBlob(ctx context.Context, numSymbols uint64, quorums
BinIndex: binIndex,
CumulativePayment: cumulativePayment,
}
protoPaymentHeader := pm.ConvertToProtoPaymentHeader()

return protoPaymentHeader, nil
return pm, nil
}

// TODO: PaymentCharged and SymbolsCharged copied from meterer, should be refactored
Expand Down Expand Up @@ -160,6 +159,37 @@ func (a *accountant) GetRelativeBinRecord(index uint32) *BinRecord {
return &a.binRecords[relativeIndex]
}

func (a *accountant) SetPaymentState(paymentState *disperser_rpc.GetPaymentStateReply) {
a.minNumSymbols = uint32(paymentState.PaymentGlobalParams.MinNumSymbols)
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

these will panic if any of the attributes are nil?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

yes.... added a check for all of them!
It would've been great if protobuf let me set fields to be required, but proto3 removed "required" keyword:(

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If you use getters (i.e. GetXXX), you could chain them!


a.onDemand.CumulativePayment = new(big.Int).SetBytes(paymentState.OnchainCumulativePayment)
a.cumulativePayment = new(big.Int).SetBytes(paymentState.CumulativePayment)
a.pricePerSymbol = uint32(paymentState.PaymentGlobalParams.PricePerSymbol)

a.reservation.SymbolsPerSec = uint64(paymentState.PaymentGlobalParams.GlobalSymbolsPerSecond)
a.reservation.StartTimestamp = uint64(paymentState.Reservation.StartTimestamp)
a.reservation.EndTimestamp = uint64(paymentState.Reservation.EndTimestamp)
a.reservationWindow = uint32(paymentState.PaymentGlobalParams.ReservationWindow)
quorumNumbers := make([]uint8, len(paymentState.Reservation.QuorumNumbers))
for i, quorum := range paymentState.Reservation.QuorumNumbers {
quorumNumbers[i] = uint8(quorum)
}
a.reservation.QuorumNumbers = quorumNumbers
quorumSplit := make([]uint8, len(paymentState.Reservation.QuorumSplit))
for i, quorum := range paymentState.Reservation.QuorumSplit {
quorumSplit[i] = uint8(quorum)
}
a.reservation.QuorumSplit = quorumSplit
binRecords := make([]BinRecord, len(paymentState.BinRecords))
for i, record := range paymentState.BinRecords {
binRecords[i] = BinRecord{
Index: record.Index,
Usage: record.Usage,
}
}
a.binRecords = binRecords
}

// QuorumCheck eagerly returns error if the check finds a quorum number not an element of the allowed quorum numbers
func QuorumCheck(quorumNumbers []uint8, allowedNumbers []uint8) error {
if len(quorumNumbers) == 0 {
Expand Down
33 changes: 11 additions & 22 deletions api/clients/accountant_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -71,30 +71,27 @@ func TestAccountBlob_Reservation(t *testing.T) {
quorums := []uint8{0, 1}

header, err := accountant.AccountBlob(ctx, symbolLength, quorums)
metadata := core.ConvertPaymentHeader(header)

assert.NoError(t, err)
assert.Equal(t, meterer.GetBinIndex(uint64(time.Now().Unix()), reservationWindow), header.BinIndex)
assert.Equal(t, big.NewInt(0), metadata.CumulativePayment)
assert.Equal(t, big.NewInt(0), header.CumulativePayment)
assert.Equal(t, isRotation([]uint64{500, 0, 0}, mapRecordUsage(accountant.binRecords)), true)

symbolLength = uint64(700)

header, err = accountant.AccountBlob(ctx, symbolLength, quorums)
metadata = core.ConvertPaymentHeader(header)

assert.NoError(t, err)
assert.NotEqual(t, 0, header.BinIndex)
assert.Equal(t, big.NewInt(0), metadata.CumulativePayment)
assert.Equal(t, big.NewInt(0), header.CumulativePayment)
assert.Equal(t, isRotation([]uint64{1200, 0, 200}, mapRecordUsage(accountant.binRecords)), true)

// Second call should use on-demand payment
header, err = accountant.AccountBlob(ctx, 300, quorums)
metadata = core.ConvertPaymentHeader(header)

assert.NoError(t, err)
assert.Equal(t, uint32(0), header.BinIndex)
assert.Equal(t, big.NewInt(300), metadata.CumulativePayment)
assert.Equal(t, big.NewInt(300), header.CumulativePayment)
}

func TestAccountBlob_OnDemand(t *testing.T) {
Expand Down Expand Up @@ -124,10 +121,9 @@ func TestAccountBlob_OnDemand(t *testing.T) {
header, err := accountant.AccountBlob(ctx, numSymbols, quorums)
assert.NoError(t, err)

metadata := core.ConvertPaymentHeader(header)
expectedPayment := big.NewInt(int64(numSymbols * uint64(pricePerSymbol)))
assert.Equal(t, uint32(0), header.BinIndex)
assert.Equal(t, expectedPayment, metadata.CumulativePayment)
assert.Equal(t, expectedPayment, header.CumulativePayment)
assert.Equal(t, isRotation([]uint64{0, 0, 0}, mapRecordUsage(accountant.binRecords)), true)
assert.Equal(t, expectedPayment, accountant.cumulativePayment)
}
Expand Down Expand Up @@ -180,24 +176,21 @@ func TestAccountBlobCallSeries(t *testing.T) {

// First call: Use reservation
header, err := accountant.AccountBlob(ctx, 800, quorums)
metadata := core.ConvertPaymentHeader(header)
assert.NoError(t, err)
assert.Equal(t, meterer.GetBinIndex(uint64(now), reservationWindow), header.BinIndex)
assert.Equal(t, big.NewInt(0), metadata.CumulativePayment)
assert.Equal(t, big.NewInt(0), header.CumulativePayment)

// Second call: Use remaining reservation + overflow
header, err = accountant.AccountBlob(ctx, 300, quorums)
metadata = core.ConvertPaymentHeader(header)
assert.NoError(t, err)
assert.Equal(t, meterer.GetBinIndex(uint64(now), reservationWindow), header.BinIndex)
assert.Equal(t, big.NewInt(0), metadata.CumulativePayment)
assert.Equal(t, big.NewInt(0), header.CumulativePayment)

// Third call: Use on-demand
header, err = accountant.AccountBlob(ctx, 500, quorums)
metadata = core.ConvertPaymentHeader(header)
assert.NoError(t, err)
assert.Equal(t, uint32(0), header.BinIndex)
assert.Equal(t, big.NewInt(500), metadata.CumulativePayment)
assert.Equal(t, big.NewInt(500), header.CumulativePayment)

// Fourth call: Insufficient on-demand
_, err = accountant.AccountBlob(ctx, 600, quorums)
Expand Down Expand Up @@ -321,23 +314,20 @@ func TestAccountBlob_ReservationWithOneOverflow(t *testing.T) {
header, err := accountant.AccountBlob(ctx, 800, quorums)
assert.NoError(t, err)
assert.Equal(t, meterer.GetBinIndex(uint64(now), reservationWindow), header.BinIndex)
metadata := core.ConvertPaymentHeader(header)
assert.Equal(t, big.NewInt(0), metadata.CumulativePayment)
assert.Equal(t, big.NewInt(0), header.CumulativePayment)
assert.Equal(t, isRotation([]uint64{800, 0, 0}, mapRecordUsage(accountant.binRecords)), true)

// Second call: Allow one overflow
header, err = accountant.AccountBlob(ctx, 500, quorums)
assert.NoError(t, err)
metadata = core.ConvertPaymentHeader(header)
assert.Equal(t, big.NewInt(0), metadata.CumulativePayment)
assert.Equal(t, big.NewInt(0), header.CumulativePayment)
assert.Equal(t, isRotation([]uint64{1300, 0, 300}, mapRecordUsage(accountant.binRecords)), true)

// Third call: Should use on-demand payment
header, err = accountant.AccountBlob(ctx, 200, quorums)
assert.NoError(t, err)
assert.Equal(t, uint32(0), header.BinIndex)
metadata = core.ConvertPaymentHeader(header)
assert.Equal(t, big.NewInt(200), metadata.CumulativePayment)
assert.Equal(t, big.NewInt(200), header.CumulativePayment)
assert.Equal(t, isRotation([]uint64{1300, 0, 300}, mapRecordUsage(accountant.binRecords)), true)
}

Expand Down Expand Up @@ -373,8 +363,7 @@ func TestAccountBlob_ReservationOverflowReset(t *testing.T) {
header, err := accountant.AccountBlob(ctx, 500, quorums)
assert.NoError(t, err)
assert.Equal(t, isRotation([]uint64{1000, 0, 0}, mapRecordUsage(accountant.binRecords)), true)
metadata := core.ConvertPaymentHeader(header)
assert.Equal(t, big.NewInt(500), metadata.CumulativePayment)
assert.Equal(t, big.NewInt(500), header.CumulativePayment)

// Wait for next reservation duration
time.Sleep(time.Duration(reservationWindow) * time.Second)
Expand Down
70 changes: 51 additions & 19 deletions api/clients/disperser_client_v2.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,6 @@ package clients
import (
"context"
"fmt"
"math/big"
"sync"

"github.com/Layr-Labs/eigenda/api"
Expand All @@ -30,12 +29,13 @@ type DisperserClientV2 interface {
}

type disperserClientV2 struct {
config *DisperserClientV2Config
signer corev2.BlobRequestSigner
initOnce sync.Once
conn *grpc.ClientConn
client disperser_rpc.DisperserClient
prover encoding.Prover
config *DisperserClientV2Config
signer corev2.BlobRequestSigner
initOnce sync.Once
conn *grpc.ClientConn
client disperser_rpc.DisperserClient
prover encoding.Prover
accountant *accountant
}

var _ DisperserClientV2 = &disperserClientV2{}
Expand All @@ -60,7 +60,7 @@ var _ DisperserClientV2 = &disperserClientV2{}
//
// // Subsequent calls will use the existing connection
// status2, blobKey2, err := client.DisperseBlob(ctx, data, blobHeader)
func NewDisperserClientV2(config *DisperserClientV2Config, signer corev2.BlobRequestSigner, prover encoding.Prover) (*disperserClientV2, error) {
func NewDisperserClientV2(config *DisperserClientV2Config, signer corev2.BlobRequestSigner, prover encoding.Prover, accountant *accountant) (*disperserClientV2, error) {
if config == nil {
return nil, api.NewErrorInvalidArg("config must be provided")
}
Expand All @@ -75,13 +75,25 @@ func NewDisperserClientV2(config *DisperserClientV2Config, signer corev2.BlobReq
}

return &disperserClientV2{
config: config,
signer: signer,
prover: prover,
config: config,
signer: signer,
prover: prover,
accountant: accountant,
// conn and client are initialized lazily
}, nil
}

// PopulateAccountant populates the accountant with the payment state from the disperser.
// This function is required to be called before using the accountant. Perhaps rename to Start()?
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This function isn't required if a valid accountant is passed in the constructor, right?
If you want to require this method, we can remove accountant as an argument from the constructor.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Oh yeah, will update the comment

func (c *disperserClientV2) PopulateAccountant(ctx context.Context) error {
paymentState, err := c.GetPaymentState(ctx)
if err != nil {
return fmt.Errorf("error getting payment state for initializing accountant: %w", err)
}
c.accountant.SetPaymentState(paymentState)
return nil
}

// Close closes the grpc connection to the disperser server.
// It is thread safe and can be called multiple times.
func (c *disperserClientV2) Close() error {
Expand Down Expand Up @@ -109,15 +121,11 @@ func (c *disperserClientV2) DisperseBlob(
return nil, [32]byte{}, api.NewErrorInternal("uninitialized signer for authenticated dispersal")
}

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

check if c.accountant is nil?
we can instruct how to initialize accountant in the error message

var payment core.PaymentMetadata
accountId, err := c.signer.GetAccountID()
symbolLength := encoding.GetBlobLengthPowerOf2(uint(len(data)))
payment, err := c.accountant.AccountBlob(ctx, uint64(symbolLength), quorums)
if err != nil {
return nil, [32]byte{}, api.NewErrorInvalidArg(fmt.Sprintf("please configure signer key if you want to use authenticated endpoint %v", err))
return nil, [32]byte{}, fmt.Errorf("error accounting blob: %w", err)
}
payment.AccountID = accountId
// TODO: add payment metadata
payment.BinIndex = 0
payment.CumulativePayment = big.NewInt(0)

if len(quorums) == 0 {
return nil, [32]byte{}, api.NewErrorInvalidArg("quorum numbers must be provided")
Expand Down Expand Up @@ -160,7 +168,7 @@ func (c *disperserClientV2) DisperseBlob(
BlobVersion: blobVersion,
BlobCommitments: blobCommitments,
QuorumNumbers: quorums,
PaymentMetadata: payment,
PaymentMetadata: *payment,
}
sig, err := c.signer.SignBlobRequest(blobHeader)
if err != nil {
Expand Down Expand Up @@ -202,6 +210,30 @@ func (c *disperserClientV2) GetBlobStatus(ctx context.Context, blobKey corev2.Bl
return c.client.GetBlobStatus(ctx, request)
}

// GetPaymentState returns the payment state of the disperser client
func (c *disperserClientV2) GetPaymentState(ctx context.Context) (*disperser_rpc.GetPaymentStateReply, error) {
err := c.initOnceGrpcConnection()
if err != nil {
return nil, api.NewErrorInternal(err.Error())
}

accountID, err := c.signer.GetAccountID()
if err != nil {
return nil, fmt.Errorf("error getting signer's account ID: %w", err)
}

signature, err := c.signer.SignPaymentStateRequest()
if err != nil {
return nil, fmt.Errorf("error signing payment state request: %w", err)
}

request := &disperser_rpc.GetPaymentStateRequest{
AccountId: accountID,
Signature: signature,
}
return c.client.GetPaymentState(ctx, request)
}

// GetBlobCommitment is a utility method that calculates commitment for a blob payload.
// While the blob commitment can be calculated by anyone, it requires SRS points to
// be loaded. For service that does not have access to SRS points, this method can be
Expand Down
5 changes: 5 additions & 0 deletions disperser/apiserver/server_v2.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -255,6 +255,10 @@ func (s *DispersalServerV2) GetPaymentState(ctx context.Context, req *pb.GetPaym
for i, v := range reservation.QuorumNumbers {
quorumNumbers[i] = uint32(v)
}
quorumSplit := make([]uint32, len(reservation.QuorumSplit))
for i, v := range reservation.QuorumSplit {
quorumSplit[i] = uint32(v)
}
// build reply
reply := &pb.GetPaymentStateReply{
PaymentGlobalParams: &paymentGlobalParams,
Expand All @@ -264,6 +268,7 @@ func (s *DispersalServerV2) GetPaymentState(ctx context.Context, req *pb.GetPaym
StartTimestamp: uint32(reservation.StartTimestamp),
EndTimestamp: uint32(reservation.EndTimestamp),
QuorumNumbers: quorumNumbers,
QuorumSplit: quorumSplit,
},
CumulativePayment: largestCumulativePayment.Bytes(),
OnchainCumulativePayment: onDemandPayment.CumulativePayment.Bytes(),
Expand Down
Loading