Skip to content
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

[v2] disperser client payments api #928

Merged
merged 6 commits into from
Dec 6, 2024
Merged

[v2] disperser client payments api #928

Show file tree
Hide file tree
Changes from 5 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
62 changes: 57 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)
}
samlaf marked this conversation as resolved.
Show resolved Hide resolved
samlaf marked this conversation as resolved.
Show resolved Hide resolved

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,59 @@ func (a *accountant) GetRelativeBinRecord(index uint32) *BinRecord {
return &a.binRecords[relativeIndex]
}

func (a *accountant) SetPaymentState(paymentState *disperser_rpc.GetPaymentStateReply) error {
Copy link
Contributor

Choose a reason for hiding this comment

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

This feels too coupled to me. A Public setter that is not part of the interface and it set by some response from another client. As discussed in other threads, could we not combine accountant and disperser_client more smoothly? Does accountant really need to be an independent struct?

Copy link
Contributor Author

Choose a reason for hiding this comment

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

We wanted to keep accounting flow separate from the client, but arguably we can combine them into a bigger struct. What do you think @ian-shim ?

Copy link
Contributor

Choose a reason for hiding this comment

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

I think this would improve the UX quite a bit. You cannot disperse a blob without payment metadata and you probably don't need accountant functionality other than for the purpose of dispersing a blob, so coupling them more explicitly makes sense imo

Copy link
Contributor

Choose a reason for hiding this comment

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

Makes sense! I would also not try to future-proof this design too much (to a case where there are multiple dispersers and you need some super complex accountant that manages multiple dispersers). We should try to make the simple case simple, true to golang's philosophy, and people who need complex integrations can take on the complexity hit around our simple API.

Another possibility is to have 2 constructors: one where we inject the accountant, and the other that initializes the accountant by calling the disperser?

Copy link
Contributor Author

Choose a reason for hiding this comment

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

Hmm all one needs to do is to call the current constructor and then call PopulateAccountant, so adding a separate constructor doesn't seem that beneficial to me. I won't mind adding it, probably in the follow-up PR that focus on integrations adaptions (maybe something like NewDisperserClientV2AutoPay?

if paymentState == nil {
return fmt.Errorf("payment state cannot be nil")
} else if paymentState.GetPaymentGlobalParams() == nil {
return fmt.Errorf("payment global params cannot be nil")
} else if paymentState.GetOnchainCumulativePayment() == nil {
return fmt.Errorf("onchain cumulative payment cannot be nil")
} else if paymentState.GetCumulativePayment() == nil {
return fmt.Errorf("cumulative payment cannot be nil")
} else if paymentState.GetReservation() == nil {
return fmt.Errorf("reservation cannot be nil")
} else if paymentState.GetReservation().GetQuorumNumbers() == nil {
return fmt.Errorf("reservation quorum numbers cannot be nil")
} else if paymentState.GetReservation().GetQuorumSplit() == nil {
return fmt.Errorf("reservation quorum split cannot be nil")
} else if paymentState.GetBinRecords() == nil {
return fmt.Errorf("bin records cannot be nil")
}

a.minNumSymbols = uint32(paymentState.PaymentGlobalParams.MinNumSymbols)
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

return nil
}

// 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
76 changes: 57 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,28 @@ 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.
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)
}

err = c.accountant.SetPaymentState(paymentState)
if err != nil {
return fmt.Errorf("error setting payment state for accountant: %w", err)
}
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 All @@ -108,16 +123,15 @@ func (c *disperserClientV2) DisperseBlob(
if c.signer == nil {
return nil, [32]byte{}, api.NewErrorInternal("uninitialized signer for authenticated dispersal")
}
if c.accountant == nil {
return nil, [32]byte{}, api.NewErrorInternal("uninitialized accountant for paid dispersal; make sure to call PopulateAccountant after creating the client")
Copy link
Contributor

Choose a reason for hiding this comment

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

how about we make the call to PopulateAccountant as part of the initOnce function? This way its hidden from the user.
Is there ever a need to recall PopulateAccountant? For eg if disperser went down, or whatever other weird case? To reset the accountant basically? If so, should it be called ResetAccountantFromServer or something like that?

Copy link
Contributor Author

Choose a reason for hiding this comment

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

In this case, maybe we also move signer check to the initOnce?

There's some cases a user might recall populate accountant, I don't think it hurts to be able to reset. Can update the function name for sure

Copy link
Contributor Author

Choose a reason for hiding this comment

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

I think I'll leave this as a future refactoring

}

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 +174,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 +216,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