[v2] payments inabox (#991)

Author: hopeyen

api/clients/v2/disperser_client.go

@@ -3,7 +3,6 @@ package clients
 import (
 	"context"
 	"fmt"
-	"math/big"
 	"sync"
 
 	"github.com/Layr-Labs/eigenda/api"
@@ -139,16 +138,11 @@ func (c *disperserClient) DisperseBlob(
 		return nil, [32]byte{}, api.NewErrorInternal("uninitialized signer for authenticated dispersal")
 	}
 
-	// TODO(hopeyen): uncomment this after the accountant is implemented
-	// if c.accountant == nil {
-	// 	return nil, [32]byte{}, api.NewErrorInternal("uninitialized accountant for paid dispersal; make sure to call PopulateAccountant after creating the client")
-	// }
-
-	// symbolLength := encoding.GetBlobLengthPowerOf2(uint(len(data)))
-	// payment, err := c.accountant.AccountBlob(ctx, uint32(symbolLength), quorums, salt)
-	// if err != nil {
-	// 	return nil, [32]byte{}, fmt.Errorf("error accounting blob: %w", err)
-	// }
+	symbolLength := encoding.GetBlobLengthPowerOf2(uint(len(data)))
+	payment, err := c.accountant.AccountBlob(ctx, uint32(symbolLength), quorums, salt)
+	if err != nil {
+		return nil, [32]byte{}, fmt.Errorf("error accounting blob: %w", err)
+	}
 
 	if len(quorums) == 0 {
 		return nil, [32]byte{}, api.NewErrorInvalidArg("quorum numbers must be provided")
@@ -187,26 +181,18 @@ func (c *disperserClient) DisperseBlob(
 		}
 	}
 
-	var payment core.PaymentMetadata
-	accountId, err := c.signer.GetAccountID()
-	if err != nil {
-		return nil, [32]byte{}, api.NewErrorInvalidArg(fmt.Sprintf("please configure signer key if you want to use authenticated endpoint %v", err))
-	}
-	payment.AccountID = accountId
-	payment.ReservationPeriod = 0
-	payment.CumulativePayment = big.NewInt(0)
 	blobHeader := &corev2.BlobHeader{
 		BlobVersion:     blobVersion,
 		BlobCommitments: blobCommitments,
 		QuorumNumbers:   quorums,
-		PaymentMetadata: payment,
-	}
-	// TODO(hopeyen): uncomment this and replace the payment metadata for authentication
-	// sig, err := c.signer.SignBlobRequest(blobHeader)
-	// if err != nil {
-	// 	return nil, [32]byte{}, fmt.Errorf("error signing blob request: %w", err)
-	// }
-	// blobHeader.Signature = sig
+		PaymentMetadata: *payment,
+	}
+
+	sig, err := c.signer.SignBlobRequest(blobHeader)
+	if err != nil {
+		return nil, [32]byte{}, fmt.Errorf("error signing blob request: %w", err)
+	}
+	blobHeader.Signature = sig
 	blobHeaderProto, err := blobHeader.ToProtobuf()
 	if err != nil {
 		return nil, [32]byte{}, fmt.Errorf("error converting blob header to protobuf: %w", err)

contracts/script/SetUpEigenDA.s.sol

@@ -10,6 +10,8 @@ import {StakeRegistry} from "eigenlayer-middleware/StakeRegistry.sol";
 import {IIndexRegistry} from "eigenlayer-middleware/interfaces/IIndexRegistry.sol";
 
 import {EigenDAServiceManager} from "../src/core/EigenDAServiceManager.sol";
+import {PaymentVault} from "../src/payments/PaymentVault.sol";
+import {IPaymentVault} from "../src/interfaces/IPaymentVault.sol";
 import {EigenDAHasher} from "../src/libraries/EigenDAHasher.sol";
 import {EigenDADeployer} from "./EigenDADeployer.s.sol";
 import {EigenLayerUtils} from "./EigenLayerUtils.s.sol";
@@ -19,6 +21,21 @@ import "forge-std/Test.sol";
 import "forge-std/Script.sol";
 import "forge-std/StdJson.sol";
 
+
+// Helper function to create single-element arrays
+function toArray(address element) pure returns (address[] memory) {
+    address[] memory arr = new address[](1);
+    arr[0] = element;
+    return arr;
+}
+
+function toArray(uint256 element) pure returns (uint256[] memory) {
+    uint256[] memory arr = new uint256[](1);
+    arr[0] = element;
+    return arr;
+}
+
+
 // # To load the variables in the .env file
 // source .env
 // # To deploy and verify our contract
@@ -112,8 +129,7 @@ contract SetupEigenDA is EigenDADeployer, EigenLayerUtils {
         }
 
         vm.startBroadcast();
-
-        // Allocate eth to stakers and operators
+        // Allocate eth to stakers, operators, dispserser clients
         _allocate(
             IERC20(address(0)),
             stakers,
@@ -156,6 +172,22 @@ contract SetupEigenDA is EigenDADeployer, EigenLayerUtils {
             delegation.registerAsOperator(IDelegationManager.OperatorDetails(earningsReceiver, delegationApprover, stakerOptOutWindowBlocks), metadataURI);
         }
 
+
+        // Register Reservations for client as the eigenDACommunityMultisig
+        IPaymentVault.Reservation memory reservation = IPaymentVault.Reservation({
+            symbolsPerSecond: 452198,
+            startTimestamp: uint64(block.timestamp),
+            endTimestamp: uint64(block.timestamp + 1000000000),
+            quorumNumbers: hex"0001",
+            quorumSplits: hex"3232"
+        });
+        address clientAddress = address(0x641691973c98dFe68b07Ee3613E270406285DFE8);
+        vm.startBroadcast(msg.sender);
+        paymentVault.setReservation(clientAddress, reservation);
+        // Deposit OnDemand 
+        paymentVault.depositOnDemand{value: 0.1 ether}(clientAddress);
+        vm.stopBroadcast();
+
         // Deposit stakers into EigenLayer and delegate to operators
         for (uint256 i = 0; i < stakerPrivateKeys.length; i++) {
             vm.startBroadcast(stakerPrivateKeys[i]);

core/meterer/onchain_state_test.go

@@ -27,7 +27,7 @@ var (
 func TestRefreshOnchainPaymentState(t *testing.T) {
 	mockState := &mock.MockOnchainPaymentState{}
 	ctx := context.Background()
-	mockState.On("RefreshOnchainPaymentState", testifymock.Anything, testifymock.Anything).Return(nil)
+	mockState.On("RefreshOnchainPaymentState", testifymock.Anything).Return(nil)
 
 	err := mockState.RefreshOnchainPaymentState(ctx)
 	assert.NoError(t, err)

disperser/apiserver/disperse_blob_v2.go

@@ -4,10 +4,12 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"math/big"
 	"time"
 
 	"github.com/Layr-Labs/eigenda/api"
 	pb "github.com/Layr-Labs/eigenda/api/grpc/disperser/v2"
+	"github.com/Layr-Labs/eigenda/core"
 	corev2 "github.com/Layr-Labs/eigenda/core/v2"
 	"github.com/Layr-Labs/eigenda/disperser/common"
 	dispv2 "github.com/Layr-Labs/eigenda/disperser/common/v2"
@@ -107,6 +109,19 @@ func (s *DispersalServerV2) validateDispersalRequest(ctx context.Context, req *p
 		return api.NewErrorInvalidArg("blob header must contain commitments")
 	}
 
+	blobHeader, err := corev2.BlobHeaderFromProtobuf(blobHeaderProto)
+	if err != nil {
+		return api.NewErrorInvalidArg(fmt.Sprintf("invalid blob header: %s", err.Error()))
+	}
+
+	if blobHeader.PaymentMetadata == (core.PaymentMetadata{}) {
+		return api.NewErrorInvalidArg("payment metadata is required")
+	}
+
+	if len(blobHeader.PaymentMetadata.AccountID) == 0 || blobHeader.PaymentMetadata.ReservationPeriod == 0 || blobHeader.PaymentMetadata.CumulativePayment == nil {
+		return api.NewErrorInvalidArg("invalid payment metadata")
+	}
+
 	if len(blobHeaderProto.GetQuorumNumbers()) == 0 {
 		return api.NewErrorInvalidArg("blob header must contain at least one quorum number")
 	}
@@ -122,7 +137,7 @@ func (s *DispersalServerV2) validateDispersalRequest(ctx context.Context, req *p
 	}
 
 	// validate every 32 bytes is a valid field element
-	_, err := rs.ToFrArray(data)
+	_, err = rs.ToFrArray(data)
 	if err != nil {
 		s.logger.Error("failed to convert a 32bytes as a field element", "err", err)
 		return api.NewErrorInvalidArg("encountered an error to convert a 32-bytes into a valid field element, please use the correct format where every 32bytes(big-endian) is less than 21888242871839275222246405745257275088548364400416034343698204186575808495617")
@@ -132,38 +147,25 @@ func (s *DispersalServerV2) validateDispersalRequest(ctx context.Context, req *p
 		return api.NewErrorInvalidArg(fmt.Sprintf("invalid blob version %d; valid blob versions are: %v", blobHeaderProto.GetVersion(), onchainState.BlobVersionParameters.Keys()))
 	}
 
-	blobHeader, err := corev2.BlobHeaderFromProtobuf(blobHeaderProto)
-	if err != nil {
-		return api.NewErrorInvalidArg(fmt.Sprintf("invalid blob header: %s", err.Error()))
+	if err = s.authenticator.AuthenticateBlobRequest(blobHeader); err != nil {
+		return api.NewErrorInvalidArg(fmt.Sprintf("authentication failed: %s", err.Error()))
 	}
-	// TODO(ian-shim): enable this check for authentication
-	// if blobHeader.PaymentMetadata == nil {
-	// 	return api.NewErrorInvalidArg("payment metadata is required")
-	// }
-	// if err = s.authenticator.AuthenticateBlobRequest(blobHeader); err != nil {
-	// 	return api.NewErrorInvalidArg(fmt.Sprintf("authentication failed: %s", err.Error()))
-	// }
-
-	// TODO(ian-shim): enable this check when we have payment metadata + authentication in disperser client
-	// if len(blobHeader.PaymentMetadata.AccountID) == 0 || (blobHeader.PaymentMetadata.ReservationPeriod == 0 && blobHeader.PaymentMetadata.CumulativePayment == nil) {
-	// 	return api.NewErrorInvalidArg("invalid payment metadata")
-	// }
 
 	// handle payments and check rate limits
-	// reservationPeriod := blobHeaderProto.GetPaymentHeader().GetReservationPeriod()
-	// cumulativePayment := new(big.Int).SetBytes(blobHeaderProto.GetPaymentHeader().GetCumulativePayment())
-	// accountID := blobHeaderProto.GetPaymentHeader().GetAccountId()
-
-	// paymentHeader := core.PaymentMetadata{
-	// 	AccountID:         accountID,
-	// 	ReservationPeriod:          reservationPeriod,
-	// 	CumulativePayment: cumulativePayment,
-	// }
-
-	// err := s.meterer.MeterRequest(ctx, paymentHeader, blobLength, blobHeader.QuorumNumbers)
-	// if err != nil {
-	// 	return api.NewErrorResourceExhausted(err.Error())
-	// }
+	reservationPeriod := blobHeaderProto.GetPaymentHeader().GetReservationPeriod()
+	cumulativePayment := new(big.Int).SetBytes(blobHeaderProto.GetPaymentHeader().GetCumulativePayment())
+	accountID := blobHeaderProto.GetPaymentHeader().GetAccountId()
+
+	paymentHeader := core.PaymentMetadata{
+		AccountID:         accountID,
+		ReservationPeriod: reservationPeriod,
+		CumulativePayment: cumulativePayment,
+	}
+
+	err = s.meterer.MeterRequest(ctx, paymentHeader, blobLength, blobHeader.QuorumNumbers)
+	if err != nil {
+		return api.NewErrorResourceExhausted(err.Error())
+	}
 
 	commitments, err := s.prover.GetCommitmentsForPaddedLength(data)
 	if err != nil {

disperser/apiserver/server_v2.go

@@ -269,6 +269,7 @@ func (s *DispersalServerV2) GetPaymentState(ctx context.Context, req *pb.GetPaym
 
 	// validate the signature
 	if err := s.authenticator.AuthenticatePaymentStateRequest(req.GetSignature(), req.GetAccountId()); err != nil {
+		s.logger.Debug("failed to validate signature", "err", err, "accountID", accountID)
 		return nil, api.NewErrorInvalidArg(fmt.Sprintf("authentication failed: %s", err.Error()))
 	}
 	// on-chain global payment parameters

disperser/apiserver/server_v2_test.go

@@ -107,13 +107,14 @@ func TestV2DisperseBlob(t *testing.T) {
 	assert.Greater(t, blobMetadata.RequestedAt, uint64(now.UnixNano()))
 	assert.Equal(t, blobMetadata.RequestedAt, blobMetadata.UpdatedAt)
 
-	// Try dispersing the same blob
+	// Try dispersing the same blob; if payment is different, blob will be considered as a differernt blob
+	// payment will cause failure before commitment check
 	reply, err = c.DispersalServerV2.DisperseBlob(ctx, &pbv2.DisperseBlobRequest{
 		Data:       data,
 		BlobHeader: blobHeaderProto,
 	})
 	assert.Nil(t, reply)
-	assert.ErrorContains(t, err, "blob already exists")
+	assert.ErrorContains(t, err, "payment already exists")
 }
 
 func TestV2DisperseBlobRequestValidation(t *testing.T) {
@@ -212,9 +213,7 @@ func TestV2DisperseBlobRequestValidation(t *testing.T) {
 		Data:       data,
 		BlobHeader: invalidReqProto,
 	})
-	// TODO(hopeyen); re-enable this validation after adding signature verification
-	// assert.ErrorContains(t, err, "authentication failed")
-	assert.NoError(t, err)
+	assert.ErrorContains(t, err, "authentication failed")
 
 	// request with invalid payment metadata
 	invalidReqProto = &pbcommonv2.BlobHeader{
@@ -237,9 +236,7 @@ func TestV2DisperseBlobRequestValidation(t *testing.T) {
 		Data:       data,
 		BlobHeader: invalidReqProto,
 	})
-	// TODO(ian-shim): re-enable this validation after fixing the payment metadata validation
-	// assert.ErrorContains(t, err, "invalid payment metadata")
-	assert.NoError(t, err)
+	assert.ErrorContains(t, err, "invalid payment metadata")
 
 	// request with invalid commitment
 	invalidCommitment := commitmentProto

inabox/deploy/localstack.go

@@ -135,30 +135,31 @@ func DeployResources(
 		return err
 	}
 
+	fmt.Println("Creating v2 tables")
 	if v2MetadataTableName != "" {
 		// Create v2 metadata table
 		_, err = test_utils.CreateTable(context.Background(), cfg, v2MetadataTableName, blobstorev2.GenerateTableSchema(v2MetadataTableName, 10, 10))
 		if err != nil {
 			return err
 		}
-	}
 
-	v2PaymentName := "e2e_v2_"
-	// create payment related tables
-	err = meterer.CreateReservationTable(cfg, v2PaymentName+"reservation")
-	if err != nil {
-		fmt.Println("err", err)
-		return err
-	}
-	err = meterer.CreateOnDemandTable(cfg, v2PaymentName+"ondemand")
-	if err != nil {
-		fmt.Println("err", err)
-		return err
-	}
-	err = meterer.CreateGlobalReservationTable(cfg, v2PaymentName+"global_reservation")
-	if err != nil {
-		fmt.Println("err", err)
-		return err
+		v2PaymentName := "e2e_v2_"
+		// create payment related tables
+		err = meterer.CreateReservationTable(cfg, v2PaymentName+"reservation")
+		if err != nil {
+			fmt.Println("err", err)
+			return err
+		}
+		err = meterer.CreateOnDemandTable(cfg, v2PaymentName+"ondemand")
+		if err != nil {
+			fmt.Println("err", err)
+			return err
+		}
+		err = meterer.CreateGlobalReservationTable(cfg, v2PaymentName+"global_reservation")
+		if err != nil {
+			fmt.Println("err", err)
+			return err
+		}
 	}
 
 	return err