disperser meterer for payments

common/aws/dynamodb/client.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"math"
+	"strconv"
 	"sync"
 
 	commonaws "github.com/Layr-Labs/eigenda/common/aws"
@@ -156,6 +157,51 @@ func (c *Client) UpdateItem(ctx context.Context, tableName string, key Key, item
 	return resp.Attributes, err
 }
 
+func (c *Client) UpdateItemIncrement(ctx context.Context, tableName string, key Key, item Item) (Item, error) {
+	update := expression.UpdateBuilder{}
+	for itemKey, itemValue := range item {
+		if _, ok := key[itemKey]; ok {
+			// Cannot update the key
+			continue
+		}
+		// fmt.Println("updating item", itemKey, itemValue)
+		// ADD numeric values
+		if n, ok := itemValue.(*types.AttributeValueMemberN); ok {
+			// update = update.Add(expression.Name(itemKey), expression.Value(n.Value))
+			// update = update.Add(expression.Name(itemKey), expression.Value(n.Value))
+			f, _ := strconv.ParseFloat(n.Value, 64)
+			update = update.Add(expression.Name(itemKey), expression.Value(aws.Float64(f)))
+
+		} else {
+			// For non-numeric values, use SET as before
+			update = update.Set(expression.Name(itemKey), expression.Value(itemValue))
+		}
+	}
+
+	expr, err := expression.NewBuilder().WithUpdate(update).Build()
+	if err != nil {
+		return nil, err
+	}
+
+	fmt.Println("update item increment", expr.Update())
+	resp, err := c.dynamoClient.UpdateItem(ctx, &dynamodb.UpdateItemInput{
+		TableName:                 aws.String(tableName),
+		Key:                       key,
+		ExpressionAttributeNames:  expr.Names(),
+		ExpressionAttributeValues: expr.Values(),
+		UpdateExpression:          expr.Update(),
+		ReturnValues:              types.ReturnValueUpdatedNew,
+	})
+	if err != nil {
+		fmt.Println("error updating item", err)
+		return nil, err
+	}
+
+	fmt.Println("update item increment", resp.Attributes)
+
+	return resp.Attributes, nil
+}
+
 func (c *Client) GetItem(ctx context.Context, tableName string, key Key) (Item, error) {
 	resp, err := c.dynamoClient.GetItem(ctx, &dynamodb.GetItemInput{Key: key, TableName: aws.String(tableName)})
 	if err != nil {

core/eth/tx.go

@@ -979,3 +979,99 @@ func bitmapToBytesArray(bitmap *big.Int) []byte {
 	}
 	return bytesArray
 }
+
+// // GetActiveReservations retrieves all active reservations from the payment contract
+// func (t *Transactor) GetActiveReservations(ctx context.Context, blockNumber uint) (map[string]*core.ActiveReservation, error) {
+// 	opts := &bind.CallOpts{
+// 		Context:     ctx,
+// 		BlockNumber: big.NewInt(int64(blockNumber)),
+// 	}
+
+// 	// Assuming the contract has a method to get all active reservation accounts
+// 	accounts, err := t.Bindings.PaymentContract.GetActiveReservationAccounts(opts)
+// 	if err != nil {
+// 		return nil, err
+// 	}
+
+// 	reservations := make(map[string]*core.ActiveReservation)
+// 	for _, account := range accounts {
+// 		reservation, err := t.Bindings.PaymentContract.GetReservation(opts, account)
+// 		if err != nil {
+// 			return nil, err
+// 		}
+// 		reservations[account.String()] = &core.ActiveReservation{
+// 			DataRate:    reservation.DataRate,
+// 			StartEpoch:  reservation.StartEpoch,
+// 			EndEpoch:    reservation.EndEpoch,
+// 			QuorumSplit: reservation.QuorumSplit,
+// 		}
+// 	}
+
+// 	return reservations, nil
+// }
+
+// // GetActiveReservationByAccount retrieves the active reservation for a specific account
+// func (t *Transactor) GetActiveReservationByAccount(ctx context.Context, blockNumber uint, accountID string) (*core.ActiveReservation, error) {
+// 	opts := &bind.CallOpts{
+// 		Context:     ctx,
+// 		BlockNumber: big.NewInt(int64(blockNumber)),
+// 	}
+
+// 	account := common.HexToAddress(accountID)
+// 	reservation, err := t.Bindings.PaymentContract.GetReservation(opts, account)
+// 	if err != nil {
+// 		return nil, err
+// 	}
+
+// 	return &core.ActiveReservation{
+// 		DataRate:    reservation.DataRate,
+// 		StartEpoch:  reservation.StartEpoch,
+// 		EndEpoch:    reservation.EndEpoch,
+// 		QuorumSplit: reservation.QuorumSplit,
+// 	}, nil
+// }
+
+// // GetOnDemandPayments retrieves all on-demand payments from the payment contract
+// func (t *Transactor) GetOnDemandPayments(ctx context.Context, blockNumber uint) (map[string]*core.OnDemandPayment, error) {
+// 	opts := &bind.CallOpts{
+// 		Context:     ctx,
+// 		BlockNumber: big.NewInt(int64(blockNumber)),
+// 	}
+
+// 	// Assuming the contract has a method to get all on-demand payment accounts
+// 	accounts, err := t.Bindings.PaymentContract.GetOnDemandPaymentAccounts(opts)
+// 	if err != nil {
+// 		return nil, err
+// 	}
+
+// 	payments := make(map[string]*core.OnDemandPayment)
+// 	for _, account := range accounts {
+// 		payment, err := t.Bindings.PaymentContract.GetOnDemandPayment(opts, account)
+// 		if err != nil {
+// 			return nil, err
+// 		}
+// 		payments[account.String()] = &core.OnDemandPayment{
+// 			CumulativePayment: core.TokenAmount(payment.CumulativePayment.Uint64()),
+// 		}
+// 	}
+
+// 	return payments, nil
+// }
+
+// // GetOnDemandPaymentByAccount retrieves the on-demand payment for a specific account
+// func (t *Transactor) GetOnDemandPaymentByAccount(ctx context.Context, blockNumber uint, accountID string) (*core.OnDemandPayment, error) {
+// 	opts := &bind.CallOpts{
+// 		Context:     ctx,
+// 		BlockNumber: big.NewInt(int64(blockNumber)),
+// 	}
+
+// 	account := common.HexToAddress(accountID)
+// 	payment, err := t.Bindings.PaymentContract.GetOnDemandPayment(opts, account)
+// 	if err != nil {
+// 		return nil, err
+// 	}
+
+// 	return &core.OnDemandPayment{
+// 		CumulativePayment: core.TokenAmount(payment.CumulativePayment.Uint64()),
+// 	}, nil
+// }

core/types.go

@@ -0,0 +1,70 @@
+package core
+
+import (
+	"math/big"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/common/math"
+	"github.com/ethereum/go-ethereum/signer/core/apitypes"
+)
+
+// EIP712Domain represents the EIP-712 domain for our blob headers
+var EIP712Domain = apitypes.TypedDataDomain{
+	Name:              "EigenDA",
+	Version:           "1",
+	ChainId:           (*math.HexOrDecimal256)(big.NewInt(17000)),
+	VerifyingContract: common.HexToAddress("0x1234000000000000000000000000000000000000").Hex(),
+}
+
+// Protocol defines parameters: epoch length and rate-limit window interval
+type ActiveReservation struct {
+	dataRate    uint32 // bandwith being reserved
+	startEpoch  uint32 // index of epoch where reservation begins
+	endEpoch    uint32 // index of epoch where reservation ends
+	quorumSplit []byte // each byte is a percentage at the corresponding quorum index
+}
+
+// Protocol defines parameters: FixedFeePerByte; fine to leave global rate-limit offchain atm
+type OnDemandPayment struct {
+	amountDeposited big.Int
+	// amountCollected big.Int
+}
+
+// // Create the typed data for EIP-712 signature verification
+// typedData := apitypes.TypedData{
+// 	Types: apitypes.Types{
+// 		"EIP712Domain": []apitypes.Type{
+// 			{Name: "name", Type: "string"},
+// 			{Name: "version", Type: "string"},
+// 			{Name: "chainId", Type: "uint256"},
+// 			{Name: "verifyingContract", Type: "address"},
+// 		},
+// 		"BlobHeader": []apitypes.Type{
+// 			{Name: "version", Type: "uint32"},
+// 			{Name: "accountID", Type: "string"},
+// 			{Name: "nonce", Type: "uint32"},
+// 			{Name: "binIndex", Type: "uint32"},
+// 			{Name: "cumulativePayment", Type: "uint64"},
+// 			{Name: "commitment", Type: "bytes"},
+// 			{Name: "dataLength", Type: "uint32"},
+// 			{Name: "blobQuorumParams", Type: "BlobQuorumParam[]"},
+// 		},
+// 		"BlobQuorumParam": []apitypes.Type{
+// 			{Name: "quorumID", Type: "uint8"},
+// 			{Name: "adversaryThreshold", Type: "uint32"},
+// 			{Name: "quorumThreshold", Type: "uint32"},
+// 		},
+// 	},
+// 	Domain:      EIP712Domain,
+// 	PrimaryType: "BlobHeader",
+// 	Message: apitypes.TypedDataMessage{
+// 		"version":           header.Version,
+// 		"accountID":         header.AccountID,
+// 		"nonce":             header.Nonce,
+// 		"binIndex":          header.BinIndex,
+// 		"cumulativePayment": header.CumulativePayment,
+// 		"commitment":        header.Commitment.Bytes(),
+// 		"dataLength":        header.DataLength,
+// 		"blobQuorumParams":  header.BlobQuorumParams,
+// 	},
+// }

disperser/apiserver/server.go

@@ -4,14 +4,15 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"google.golang.org/grpc/status"
 	"math/rand"
 	"net"
 	"slices"
 	"strings"
 	"sync"
 	"time"
 
+	"google.golang.org/grpc/status"
+
 	"github.com/Layr-Labs/eigenda/api"
 	commonpb "github.com/Layr-Labs/eigenda/api/grpc/common"
 	pb "github.com/Layr-Labs/eigenda/api/grpc/disperser"
@@ -20,6 +21,7 @@ import (
 	"github.com/Layr-Labs/eigenda/core"
 	"github.com/Layr-Labs/eigenda/core/auth"
 	"github.com/Layr-Labs/eigenda/disperser"
+	"github.com/Layr-Labs/eigenda/disperser/meterer"
 	"github.com/Layr-Labs/eigenda/encoding"
 	"github.com/Layr-Labs/eigenda/encoding/rs"
 	"github.com/Layr-Labs/eigensdk-go/logging"
@@ -44,6 +46,7 @@ type DispersalServer struct {
 	tx           core.Transactor
 	quorumConfig QuorumConfig
 
+	meterer       *meterer.Meterer
 	ratelimiter   common.RateLimiter
 	authenticator core.BlobRequestAuthenticator
 
@@ -69,6 +72,7 @@ func NewDispersalServer(
 	tx core.Transactor,
 	_logger logging.Logger,
 	metrics *disperser.Metrics,
+	meterer *meterer.Meterer,
 	ratelimiter common.RateLimiter,
 	rateConfig RateConfig,
 	maxBlobSize int,
@@ -90,6 +94,7 @@ func NewDispersalServer(
 		tx:            tx,
 		metrics:       metrics,
 		logger:        logger,
+		meterer:       meterer,
 		ratelimiter:   ratelimiter,
 		authenticator: authenticator,
 		mu:            &sync.RWMutex{},
@@ -271,7 +276,15 @@ func (s *DispersalServer) disperseBlob(ctx context.Context, blob *core.Blob, aut
 
 	s.logger.Debug("received a new blob dispersal request", "authenticatedAddress", authenticatedAddress, "origin", origin, "blobSizeBytes", blobSize, "securityParams", strings.Join(securityParamsStrings, ", "))
 
-	if s.ratelimiter != nil {
+	// payments before ratelimits
+	if s.meterer != nil {
+		//TODO: blob request header needs to be updated for payments;
+		// the tests rely on a temporarily defined struct from disperser/meterer/types
+		// err := s.meterer.MeterRequest(ctx, blob.RequestHeader)
+		// if err != nil {
+		// 	return nil, err
+		// }
+	} else if s.ratelimiter != nil {
 		err := s.checkRateLimitsAndAddRatesToHeader(ctx, blob, origin, authenticatedAddress, apiMethodName)
 		if err != nil {
 			// Note checkRateLimitsAndAddRatesToHeader already updated the metrics for this error.

disperser/apiserver/server_test.go

@@ -10,10 +10,12 @@ import (
 	"testing"
 	"time"
 
+	commonaws "github.com/Layr-Labs/eigenda/common/aws"
 	"github.com/Layr-Labs/eigenda/core/auth"
 	"github.com/Layr-Labs/eigenda/core/mock"
 	"github.com/Layr-Labs/eigenda/disperser/apiserver"
 	"github.com/Layr-Labs/eigenda/disperser/common/blobstore"
+	"github.com/Layr-Labs/eigenda/disperser/meterer"
 	"github.com/Layr-Labs/eigenda/encoding"
 	"github.com/Layr-Labs/eigenda/encoding/utils/codec"
 	"github.com/Layr-Labs/eigensdk-go/logging"
@@ -645,6 +647,38 @@ func newTestServer(transactor core.Transactor) *apiserver.DispersalServer {
 	if err != nil {
 		panic("failed to create bucket store")
 	}
+	meterConfig := meterer.Config{
+		PricePerByte:         1,
+		GlobalBytesPerSecond: 1000,
+		ReservationWindow:    time.Minute,
+	}
+
+	paymentChainState := meterer.NewMockedOnchainPaymentState()
+
+	paymentChainState.InitializeOnchainPaymentState()
+
+	clientConfig := commonaws.ClientConfig{
+		Region:          "us-east-1",
+		AccessKey:       "localstack",
+		SecretAccessKey: "localstack",
+		EndpointURL:     fmt.Sprintf("http://0.0.0.0:4566"),
+	}
+
+	store, err := meterer.NewOffchainStore(
+		clientConfig,
+		"reservations",
+		"ondemand",
+		"global",
+		logger,
+	)
+	if err != nil {
+		teardown()
+		panic("failed to create offchain store")
+	}
+	meterer, err := meterer.NewMeterer(meterConfig, meterer.TimeoutConfig{}, paymentChainState, store, logger)
+	if err != nil {
+		panic("failed to create meterer")
+	}
 	ratelimiter := ratelimit.NewRateLimiter(prometheus.NewRegistry(), globalParams, bucketStore, logger)
 
 	rateConfig := apiserver.RateConfig{
@@ -701,7 +735,7 @@ func newTestServer(transactor core.Transactor) *apiserver.DispersalServer {
 	return apiserver.NewDispersalServer(disperser.ServerConfig{
 		GrpcPort:    "51001",
 		GrpcTimeout: 1 * time.Second,
-	}, queue, transactor, logger, disperser.NewMetrics(prometheus.NewRegistry(), "9001", logger), ratelimiter, rateConfig, testMaxBlobSize)
+	}, queue, transactor, logger, disperser.NewMetrics(prometheus.NewRegistry(), "9001", logger), meterer, ratelimiter, rateConfig, testMaxBlobSize)
 }
 
 func disperseBlob(t *testing.T, server *apiserver.DispersalServer, data []byte) (pb.BlobStatus, uint, []byte) {