Add configurable finalization

.env.example.holesky

@@ -18,6 +18,9 @@ EIGENDA_PROXY_SERVICE_MANAGER_ADDR=0xD4A7E1Bd8015057293f0D0A557088c286942e84b
 # Directory path to SRS tables
 # EIGENDA_PROXY_TARGET_CACHE_PATH=resources/SRSTables
 
+# The number of Ethereum blocks of confirmation that the DA briging transaction must have before it is assumed by the proxy to be final. The value of `0` indicates that the proxy should wait for weak-subjectivity finalization (12-14 minutes).
+# EIGENDA_PROXY_ETH_CONFIRMATION_DEPTH=6
+
 # Directory path to g1.point file
 # EIGENDA_PROXY_TARGET_KZG_G1_PATH=resources/g1.point
 

.env.example.mainnet

@@ -18,6 +18,9 @@ EIGENDA_PROXY_SERVICE_MANAGER_ADDR=0x870679E138bCdf293b7Ff14dD44b70FC97e12fc0
 # Directory path to SRS tables
 # EIGENDA_PROXY_TARGET_CACHE_PATH=resources/SRSTables
 
+# The number of Ethereum blocks of confirmation that the DA briging transaction must have before it is assumed by the proxy to be final. The value of `0` indicates that the proxy should wait for weak-subjectivity finalization (12-14 minutes).
+# EIGENDA_PROXY_ETH_CONFIRMATION_DEPTH=6
+
 # Directory path to g1.point file
 # EIGENDA_PROXY_TARGET_KZG_G1_PATH=resources/g1.point
 

e2e/setup.go

@@ -49,6 +49,13 @@ func CreateTestSuite(t *testing.T, useMemory bool) (TestSuite, func()) {
 		t.Fatal("ETHEREUM_RPC environment variable is not set")
 	}
 
+	var pollInterval time.Duration
+	if useMemory {
+		pollInterval = time.Second * 1
+	} else {
+		pollInterval = time.Minute * 1
+	}
+
 	log := oplog.NewLogger(os.Stdout, oplog.CLIConfig{
 		Level:  log.LevelDebug,
 		Format: oplog.FormatLogFmt,
@@ -59,7 +66,7 @@ func CreateTestSuite(t *testing.T, useMemory bool) (TestSuite, func()) {
 		ClientConfig: clients.EigenDAClientConfig{
 			RPC:                      holeskyDA,
 			StatusQueryTimeout:       time.Minute * 45,
-			StatusQueryRetryInterval: time.Second * 1,
+			StatusQueryRetryInterval: pollInterval,
 			DisableTLS:               false,
 			SignerPrivateKeyHex:      pk,
 		},
@@ -72,6 +79,7 @@ func CreateTestSuite(t *testing.T, useMemory bool) (TestSuite, func()) {
 		PutBlobEncodingVersion: 0x00,
 		MemstoreEnabled:        useMemory,
 		MemstoreBlobExpiration: 14 * 24 * time.Hour,
+		EthConfirmationDepth:   6,
 	}
 
 	store, err := server.LoadStore(

go.mod

@@ -3,7 +3,7 @@ module github.com/Layr-Labs/eigenda-proxy
 go 1.21
 
 require (
-	github.com/Layr-Labs/eigenda v0.7.2-0.20240606180508-e90cb7432ca5
+	github.com/Layr-Labs/eigenda v0.7.5-0.20240626225853-1645ffe3489e
 	github.com/consensys/gnark-crypto v0.12.1
 	github.com/ethereum-optimism/optimism v1.7.7
 	github.com/ethereum/go-ethereum v1.14.0

go.sum

@@ -20,8 +20,8 @@ github.com/DataDog/datadog-go v2.2.0+incompatible/go.mod h1:LButxg5PwREeZtORoXG3
 github.com/DataDog/datadog-go v3.2.0+incompatible/go.mod h1:LButxg5PwREeZtORoXG3tL4fMGNddJ+vMq1mwgfaqoQ=
 github.com/DataDog/zstd v1.5.2 h1:vUG4lAyuPCXO0TLbXvPv7EB7cNK1QV/luu55UHLrrn8=
 github.com/DataDog/zstd v1.5.2/go.mod h1:g4AWEaM3yOg3HYfnJ3YIawPnVdXJh9QME85blwSAmyw=
-github.com/Layr-Labs/eigenda v0.7.2-0.20240606180508-e90cb7432ca5 h1:PGcoSXnIlZYhwfrzqG1F2E/Sqc3ZGRqa5owryswax2s=
-github.com/Layr-Labs/eigenda v0.7.2-0.20240606180508-e90cb7432ca5/go.mod h1:gG5KSp5gGY0lywj6aZwaK9ZEF8eEVX4ilo679pFpvAA=
+github.com/Layr-Labs/eigenda v0.7.5-0.20240626225853-1645ffe3489e h1:vibQgDKVXvuKZKqdyvIuCx8MH1B5uOGrKITAUp3bzJg=
+github.com/Layr-Labs/eigenda v0.7.5-0.20240626225853-1645ffe3489e/go.mod h1:gG5KSp5gGY0lywj6aZwaK9ZEF8eEVX4ilo679pFpvAA=
 github.com/Layr-Labs/eigensdk-go v0.1.7-0.20240507215523-7e4891d5099a h1:L/UsJFw9M31FD/WgXTPFB0oxbq9Cu4Urea1xWPMQS7Y=
 github.com/Layr-Labs/eigensdk-go v0.1.7-0.20240507215523-7e4891d5099a/go.mod h1:OF9lmS/57MKxS0xpSpX0qHZl0SKkDRpvJIvsGvMN1y8=
 github.com/Microsoft/go-winio v0.6.1 h1:9/kr64B9VUZrLm5YYwbGtUJnMgqWVOdUAXu6Migciow=

server/config.go

@@ -19,6 +19,7 @@ const (
 	EigenDADisperserRPCFlagName          = "eigenda-disperser-rpc"
 	EthRPCFlagName                       = "eigenda-eth-rpc"
 	SvcManagerAddrFlagName               = "eigenda-svc-manager-addr"
+	EthConfirmationDepthFlagName         = "eigenda-eth-confirmation-depth"
 	StatusQueryRetryIntervalFlagName     = "eigenda-status-query-retry-interval"
 	StatusQueryTimeoutFlagName           = "eigenda-status-query-timeout"
 	DisableTlsFlagName                   = "eigenda-disable-tls"
@@ -50,8 +51,9 @@ type Config struct {
 	PutBlobEncodingVersion codecs.BlobEncodingVersion
 
 	// ETH vars
-	EthRPC         string
-	SvcManagerAddr string
+	EthRPC               string
+	SvcManagerAddr       string
+	EthConfirmationDepth uint64
 
 	// KZG vars
 	CacheDir string
@@ -111,10 +113,11 @@ func (c *Config) VerificationCfg() *verify.Config {
 	}
 
 	return &verify.Config{
-		Verify:         true,
-		RPCURL:         c.EthRPC,
-		SvcManagerAddr: c.SvcManagerAddr,
-		KzgConfig:      kzgCfg,
+		Verify:               true,
+		RPCURL:               c.EthRPC,
+		SvcManagerAddr:       c.SvcManagerAddr,
+		KzgConfig:            kzgCfg,
+		EthConfirmationDepth: c.EthConfirmationDepth,
 	}
 
 }
@@ -140,9 +143,11 @@ func ReadConfig(ctx *cli.Context) Config {
 		MaxBlobLength:          ctx.String(MaxBlobLengthFlagName),
 		SvcManagerAddr:         ctx.String(SvcManagerAddrFlagName),
 		EthRPC:                 ctx.String(EthRPCFlagName),
+		EthConfirmationDepth:   ctx.Uint64(EthConfirmationDepthFlagName),
 		MemstoreEnabled:        ctx.Bool(MemstoreFlagName),
 		MemstoreBlobExpiration: ctx.Duration(MemstoreExpirationFlagName),
 	}
+	cfg.ClientConfig.WaitForFinalization = (cfg.EthConfirmationDepth != 0)
 	return cfg
 }
 
@@ -245,6 +250,12 @@ func CLIFlags(envPrefix string) []cli.Flag {
 			Usage:   "The deployed EigenDA service manager address. The list can be found here: https://github.com/Layr-Labs/eigenlayer-middleware/?tab=readme-ov-file#current-mainnet-deployment",
 			EnvVars: prefixEnvVars("SERVICE_MANAGER_ADDR"),
 		},
+		&cli.Uint64Flag{
+			Name:    EthConfirmationDepthFlagName,
+			Usage:   "The number of Ethereum blocks of confirmation that the DA briging transaction must have before it is assumed by the proxy to be final. The value of `0` indicates that the proxy should wait for weak-subjectivity finalization (12-14 minutes).",
+			EnvVars: prefixEnvVars("ETH_CONFIRMATION_DEPTH"),
+			Value:   6,
+		},
 		&cli.BoolFlag{
 			Name:    MemstoreFlagName,
 			Usage:   "Whether to use mem-store for DA logic.",

server/eigenda_store.go

@@ -2,27 +2,40 @@ package server
 
 import (
 	"context"
+	"errors"
 	"fmt"
+	"time"
 
 	"github.com/Layr-Labs/eigenda-proxy/verify"
 	"github.com/Layr-Labs/eigenda/api/clients"
+	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/rlp"
 )
 
+type EigenDAStoreConfig struct {
+	MaxBlobSizeBytes     uint64
+	EthConfirmationDepth uint64
+
+	// The total amount of time that the client will spend waiting for EigenDA to confirm a blob
+	StatusQueryTimeout time.Duration
+}
+
 // EigenDAStore does storage interactions and verifications for blobs with DA.
 type EigenDAStore struct {
-	client           *clients.EigenDAClient
-	verifier         *verify.Verifier
-	maxBlobSizeBytes uint64
+	client   *clients.EigenDAClient
+	verifier *verify.Verifier
+	cfg      *EigenDAStoreConfig
+	log      log.Logger
 }
 
 var _ Store = (*EigenDAStore)(nil)
 
-func NewEigenDAStore(ctx context.Context, client *clients.EigenDAClient, v *verify.Verifier, maxBlobSizeBytes uint64) (*EigenDAStore, error) {
+func NewEigenDAStore(ctx context.Context, client *clients.EigenDAClient, v *verify.Verifier, log log.Logger, cfg *EigenDAStoreConfig) (*EigenDAStore, error) {
 	return &EigenDAStore{
-		client:           client,
-		verifier:         v,
-		maxBlobSizeBytes: maxBlobSizeBytes,
+		client:   client,
+		verifier: v,
+		log:      log,
+		cfg:      cfg,
 	}, nil
 }
 
@@ -50,11 +63,6 @@ func (e EigenDAStore) Get(ctx context.Context, key []byte, domain DomainType) ([
 		return nil, err
 	}
 
-	err = e.verifier.VerifyCert(&cert)
-	if err != nil {
-		return nil, err
-	}
-
 	switch domain {
 	case BinaryDomain:
 		return decodedBlob, nil
@@ -67,13 +75,16 @@ func (e EigenDAStore) Get(ctx context.Context, key []byte, domain DomainType) ([
 
 // Put disperses a blob for some pre-image and returns the associated RLP encoded certificate commit.
 func (e EigenDAStore) Put(ctx context.Context, value []byte) (comm []byte, err error) {
-	if uint64(len(value)) > e.maxBlobSizeBytes {
-		return nil, fmt.Errorf("blob is larger than max blob size: blob length %d, max blob size %d", len(value), e.maxBlobSizeBytes)
+	if uint64(len(value)) > e.cfg.MaxBlobSizeBytes {
+		return nil, fmt.Errorf("blob is larger than max blob size: blob length %d, max blob size %d", len(value), e.cfg.MaxBlobSizeBytes)
 	}
-	cert, err := e.client.PutBlob(ctx, value)
+
+	dispersalStart := time.Now()
+	blobInfo, err := e.client.PutBlob(ctx, value)
 	if err != nil {
 		return nil, err
 	}
+	cert := (*verify.Certificate)(blobInfo)
 
 	encodedBlob, err := e.client.GetCodec().EncodeBlob(value)
 	if err != nil {
@@ -84,6 +95,31 @@ func (e EigenDAStore) Put(ctx context.Context, value []byte) (comm []byte, err e
 		return nil, err
 	}
 
+	dispersalDuration := time.Since(dispersalStart)
+	remainingTimeout := e.cfg.StatusQueryTimeout - dispersalDuration
+
+	ticker := time.NewTicker(12 * time.Second)
+	defer ticker.Stop()
+	ctx, cancel := context.WithTimeout(context.Background(), remainingTimeout)
+	defer cancel()
+
+	done := false
+	for !done {
+		select {
+		case <-ctx.Done():
+			return nil, ctx.Err()
+		case <-ticker.C:
+			err = e.verifier.VerifyCert(cert)
+			if err == nil {
+				done = true
+			} else if !errors.Is(err, verify.ErrBatchMetadataHashNotFound) {
+				return nil, err
+			} else {
+				e.log.Info("Blob confirmed, waiting for sufficient confirmation depth...", "targetDepth", e.cfg.EthConfirmationDepth)
+			}
+		}
+	}
+
 	bytes, err := rlp.EncodeToBytes(cert)
 	if err != nil {
 		return nil, fmt.Errorf("failed to encode DA cert to RLP format: %w", err)

server/load_store.go

@@ -43,6 +43,11 @@ func LoadStore(cfg CLIConfig, ctx context.Context, log log.Logger) (Store, error
 		ctx,
 		client,
 		verifier,
-		maxBlobLength,
+		log,
+		&EigenDAStoreConfig{
+			MaxBlobSizeBytes:     maxBlobLength,
+			EthConfirmationDepth: cfg.EigenDAConfig.EthConfirmationDepth,
+			StatusQueryTimeout:   cfg.EigenDAConfig.ClientConfig.StatusQueryTimeout,
+		},
 	)
 }

verify/cert.go

@@ -1,20 +1,30 @@
 package verify
 
 import (
+	"bytes"
+	"context"
+	"errors"
 	"fmt"
+	"math/big"
 
 	binding "github.com/Layr-Labs/eigenda/contracts/bindings/EigenDAServiceManager"
+	"github.com/ethereum/go-ethereum/accounts/abi/bind"
 	"github.com/ethereum/go-ethereum/common"
 
 	"github.com/ethereum/go-ethereum/ethclient"
 	"github.com/ethereum/go-ethereum/log"
 	"golang.org/x/exp/slices"
 )
 
+var ErrBatchMetadataHashNotFound = errors.New("BatchMetadataHash not found for BatchId")
+
 // CertVerifier verifies the DA certificate against on-chain EigenDA contracts
 // to ensure disperser returned fields haven't been tampered with
 type CertVerifier struct {
-	manager *binding.ContractEigenDAServiceManagerCaller
+	ethConfirmationDepth uint64
+	manager              *binding.ContractEigenDAServiceManagerCaller
+	finalizedBlockClient *FinalizedBlockClient
+	ethClient            *ethclient.Client
 }
 
 func NewCertVerifier(cfg *Config, l log.Logger) (*CertVerifier, error) {
@@ -30,22 +40,33 @@ func NewCertVerifier(cfg *Config, l log.Logger) (*CertVerifier, error) {
 	}
 
 	return &CertVerifier{
-		manager: m,
+		manager:              m,
+		finalizedBlockClient: NewFinalizedBlockClient(client.Client()),
+		ethConfirmationDepth: cfg.EthConfirmationDepth,
+		ethClient:            client,
 	}, nil
 }
 
 func (cv *CertVerifier) VerifyBatch(header *binding.IEigenDAServiceManagerBatchHeader,
 	id uint32, recordHash [32]byte, blockNum uint32) error {
+	// 0 - Determine block context number
+	blockNumber, err := cv.getContextBlock()
+	if err != nil {
+		return err
+	}
+
 	// 1 - Verify batch hash
 
 	// 1.a - ensure that a batch hash can be looked up for a batch ID
-	expectedHash, err := cv.manager.BatchIdToBatchMetadataHash(nil, id)
+	expectedHash, err := cv.manager.BatchIdToBatchMetadataHash(&bind.CallOpts{BlockNumber: blockNumber}, id)
 	if err != nil {
 		return err
 	}
+	if bytes.Equal(expectedHash[:], make([]byte, 32)) {
+		return ErrBatchMetadataHashNotFound
+	}
 
 	// 1.b - ensure that hash generated from local cert matches one stored on-chain
-
 	actualHash, err := HashBatchMetadata(header, recordHash, blockNum)
 
 	if err != nil {
@@ -84,3 +105,23 @@ func (cv *CertVerifier) VerifyMerkleProof(inclusionProof []byte, root []byte, bl
 func (cv *CertVerifier) VerifyBlobParams(inclusionProof []byte, rootHash []byte, leafHash []byte, index uint64) error {
 	return nil
 }
+
+func (cv *CertVerifier) getContextBlock() (*big.Int, error) {
+	var blockNumber *big.Int
+	if cv.ethConfirmationDepth == 0 {
+		// Get the latest finalized block
+		blockHeader, err := cv.finalizedBlockClient.GetBlock(context.Background(), "finalized", false)
+		if err != nil {
+			return nil, err
+		}
+		blockNumber = blockHeader.Number()
+	} else {
+		blockHeader, err := cv.ethClient.BlockByNumber(context.Background(), nil)
+		if err != nil {
+			return nil, err
+		}
+		blockNumber = new(big.Int)
+		blockNumber.Sub(blockHeader.Number(), big.NewInt(int64(cv.ethConfirmationDepth-1)))
+	}
+	return blockNumber, nil
+}