feat: blob staleness check

We add a check to discard blobs that have been included in the rollup's batcher inbox a long time after it was included in an EigenDA batch by the eigenda disperser.

common/store.go

@@ -61,8 +61,17 @@ func StringToBackendType(s string) BackendType {
 type Store interface {
 	// Backend returns the backend type provider of the store.
 	BackendType() BackendType
-	// Verify verifies the given key-value pair.
-	Verify(ctx context.Context, key []byte, value []byte) error
+}
+
+type VerifyOptions struct {
+	// L1 block number at which the rollup batch was submitted to the batcher inbox.
+	// This is optional, and should be set to -1 to mean to not verify the reference block number distance check.
+	//
+	// Used to determine the validity of the eigenDA batch.
+	// The eigenDA batch header contains a reference block number (RBN) which is used to pin the stake of the eigenda operators at that specific blocks.
+	// The rollup batch containing the eigenDA cert is only valid if it was included within a certain number of blocks after the RBN.
+	// validity condition is: RBN < l1_inclusion_block_number < RBN + some_delta
+	RollupL1InclusionBlockNum int64
 }
 
 type GeneratedKeyStore interface {
@@ -71,6 +80,8 @@ type GeneratedKeyStore interface {
 	Get(ctx context.Context, key []byte) ([]byte, error)
 	// Put inserts the given value into the key-value data store.
 	Put(ctx context.Context, value []byte) (key []byte, err error)
+	// Verify verifies the given key-value pair.
+	Verify(ctx context.Context, key []byte, value []byte, opts VerifyOptions) error
 }
 
 type PrecomputedKeyStore interface {
@@ -79,4 +90,6 @@ type PrecomputedKeyStore interface {
 	Get(ctx context.Context, key []byte) ([]byte, error)
 	// Put inserts the given value into the key-value data store.
 	Put(ctx context.Context, key []byte, value []byte) error
+	// Verify verifies the given key-value pair.
+	Verify(ctx context.Context, key []byte, value []byte) error
 }

go.mod

@@ -12,7 +12,6 @@ require (
 	github.com/ethereum-optimism/optimism v1.9.5
 	github.com/ethereum/go-ethereum v1.14.11
 	github.com/go-redis/redis/v8 v8.11.5
-	github.com/golang/mock v1.2.0
 	github.com/gorilla/mux v1.8.0
 	github.com/joho/godotenv v1.5.1
 	github.com/minio/minio-go/v7 v7.0.80
@@ -21,6 +20,7 @@ require (
 	github.com/testcontainers/testcontainers-go/modules/minio v0.33.0
 	github.com/testcontainers/testcontainers-go/modules/redis v0.33.0
 	github.com/urfave/cli/v2 v2.27.5
+	go.uber.org/mock v0.4.0
 	golang.org/x/exp v0.0.0-20241009180824-f66d83c29e7c
 	google.golang.org/grpc v1.64.1
 )
@@ -273,7 +273,6 @@ require (
 	go.opentelemetry.io/otel/trace v1.24.0 // indirect
 	go.uber.org/dig v1.18.0 // indirect
 	go.uber.org/fx v1.22.2 // indirect
-	go.uber.org/mock v0.4.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
 	go.uber.org/zap v1.27.0 // indirect
 	golang.org/x/crypto v0.28.0 // indirect

go.sum

@@ -349,7 +349,6 @@ github.com/golang-jwt/jwt/v4 v4.5.1/go.mod h1:m21LjoU+eqJr34lmDMbreY2eSTRJ1cv77w
 github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b/go.mod h1:SBH7ygxi8pfUlaOkMMuAQtPIUF8ecWP5IEl/CR7VP2Q=
 github.com/golang/lint v0.0.0-20180702182130-06c8688daad7/go.mod h1:tluoj9z5200jBnyusfRPU2LqT6J+DAorxEvtC7LHB+E=
 github.com/golang/mock v1.1.1/go.mod h1:oTYuIxOrZwtPieC+H1uAHpcLFnEyAGVDL/k47Jfbm0A=
-github.com/golang/mock v1.2.0 h1:28o5sBqPkBsMGnC6b4MvE2TzSr5/AT4c/1fLqVGIwlk=
 github.com/golang/mock v1.2.0/go.mod h1:oTYuIxOrZwtPieC+H1uAHpcLFnEyAGVDL/k47Jfbm0A=
 github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
 github.com/golang/protobuf v1.3.1/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=

server/handlers.go

@@ -7,8 +7,10 @@ import (
 	"fmt"
 	"io"
 	"net/http"
+	"strconv"
 
 	"github.com/Layr-Labs/eigenda-proxy/commitments"
+	"github.com/Layr-Labs/eigenda-proxy/common"
 	"github.com/gorilla/mux"
 )
 
@@ -41,7 +43,7 @@ func (svr *Server) handleGetStdCommitment(w http.ResponseWriter, r *http.Request
 		return fmt.Errorf("failed to decode commitment %s: %w", rawCommitmentHex, err)
 	}
 
-	return svr.handleGetShared(r.Context(), w, commitment, commitmentMeta)
+	return svr.handleGetShared(r.Context(), w, r, commitment, commitmentMeta)
 }
 
 // handleGetOPKeccakCommitment handles the GET request for optimism keccak commitments.
@@ -67,7 +69,7 @@ func (svr *Server) handleGetOPKeccakCommitment(w http.ResponseWriter, r *http.Re
 		return fmt.Errorf("failed to decode commitment %s: %w", rawCommitmentHex, err)
 	}
 
-	return svr.handleGetShared(r.Context(), w, commitment, commitmentMeta)
+	return svr.handleGetShared(r.Context(), w, r, commitment, commitmentMeta)
 }
 
 // handleGetOPGenericCommitment handles the GET request for optimism generic commitments.
@@ -90,13 +92,24 @@ func (svr *Server) handleGetOPGenericCommitment(w http.ResponseWriter, r *http.R
 		return fmt.Errorf("failed to decode commitment %s: %w", rawCommitmentHex, err)
 	}
 
-	return svr.handleGetShared(r.Context(), w, commitment, commitmentMeta)
+	return svr.handleGetShared(r.Context(), w, r, commitment, commitmentMeta)
 }
 
-func (svr *Server) handleGetShared(ctx context.Context, w http.ResponseWriter, comm []byte, meta commitments.CommitmentMeta) error {
+func (svr *Server) handleGetShared(ctx context.Context, w http.ResponseWriter, r *http.Request, comm []byte, meta commitments.CommitmentMeta) error {
 	commitmentHex := hex.EncodeToString(comm)
 	svr.log.Info("Processing GET request", "commitment", commitmentHex, "commitmentMeta", meta)
-	input, err := svr.sm.Get(ctx, comm, meta.Mode)
+	l1InclusionBlockNum, err := parseBatchInclusionL1BlockNumQueryParam(r)
+	if err != nil {
+		err = MetaError{
+			Err:  fmt.Errorf("invalid l1_block_number: %w", err),
+			Meta: meta,
+		}
+		// the inclusion block query param is optional, but if it is provided and invalid, we return a 400 error
+		// to let the client know that they probably have a bug.
+		http.Error(w, err.Error(), http.StatusBadRequest)
+		return err
+	}
+	input, err := svr.sm.Get(ctx, comm, meta.Mode, common.VerifyOptions{RollupL1InclusionBlockNum: l1InclusionBlockNum})
 	if err != nil {
 		err = MetaError{
 			Err:  fmt.Errorf("get request failed with commitment %v: %w", commitmentHex, err),
@@ -114,6 +127,25 @@ func (svr *Server) handleGetShared(ctx context.Context, w http.ResponseWriter, c
 	return nil
 }
 
+// Parses the l1_inclusion_block_number query param from the request.
+// Happy path:
+//   - if the l1_inclusion_block_number is provided, it returns the parsed value.
+//
+// Unhappy paths:
+//   - if the l1_inclusion_block_number is not provided, it returns -1.
+//   - if the l1_inclusion_block_number is provided but isn't a valid integer, it returns an error and -1.
+func parseBatchInclusionL1BlockNumQueryParam(r *http.Request) (int64, error) {
+	l1BlockNumStr := r.URL.Query().Get("l1_inclusion_block_number")
+	if l1BlockNumStr != "" {
+		l1BlockNum, err := strconv.ParseInt(l1BlockNumStr, 10, 64)
+		if err != nil {
+			return -1, fmt.Errorf("invalid l1_inclusion_block_number: %w", err)
+		}
+		return l1BlockNum, nil
+	}
+	return -1, nil
+}
+
 // =================================================================================================
 // POST ROUTES
 // =================================================================================================

server/handlers_test.go

@@ -17,9 +17,9 @@ import (
 	"github.com/Layr-Labs/eigenda-proxy/mocks"
 	"github.com/Layr-Labs/eigenda/api"
 	"github.com/ethereum/go-ethereum/log"
-	"github.com/golang/mock/gomock"
 	"github.com/gorilla/mux"
 	"github.com/stretchr/testify/require"
+	"go.uber.org/mock/gomock"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 )
@@ -49,7 +49,7 @@ func TestHandlerGet(t *testing.T) {
 			name: "Failure - OP Keccak256 Internal Server Error",
 			url:  fmt.Sprintf("/get/0x00%s", testCommitStr),
 			mockBehavior: func() {
-				mockStorageMgr.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, fmt.Errorf("internal error"))
+				mockStorageMgr.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, fmt.Errorf("internal error"))
 			},
 			expectedCode: http.StatusInternalServerError,
 			expectedBody: "",
@@ -58,7 +58,7 @@ func TestHandlerGet(t *testing.T) {
 			name: "Success - OP Keccak256",
 			url:  fmt.Sprintf("/get/0x00%s", testCommitStr),
 			mockBehavior: func() {
-				mockStorageMgr.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any()).Return([]byte(testCommitStr), nil)
+				mockStorageMgr.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return([]byte(testCommitStr), nil)
 			},
 			expectedCode: http.StatusOK,
 			expectedBody: testCommitStr,
@@ -67,7 +67,7 @@ func TestHandlerGet(t *testing.T) {
 			name: "Failure - OP Alt-DA Internal Server Error",
 			url:  fmt.Sprintf("/get/0x010000%s", testCommitStr),
 			mockBehavior: func() {
-				mockStorageMgr.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, fmt.Errorf("internal error"))
+				mockStorageMgr.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, fmt.Errorf("internal error"))
 			},
 			expectedCode: http.StatusInternalServerError,
 			expectedBody: "",
@@ -76,7 +76,7 @@ func TestHandlerGet(t *testing.T) {
 			name: "Success - OP Alt-DA",
 			url:  fmt.Sprintf("/get/0x010000%s", testCommitStr),
 			mockBehavior: func() {
-				mockStorageMgr.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any()).Return([]byte(testCommitStr), nil)
+				mockStorageMgr.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return([]byte(testCommitStr), nil)
 			},
 			expectedCode: http.StatusOK,
 			expectedBody: testCommitStr,

server/routing_test.go

@@ -9,8 +9,8 @@ import (
 	"github.com/Layr-Labs/eigenda-proxy/metrics"
 	"github.com/Layr-Labs/eigenda-proxy/mocks"
 	"github.com/ethereum/go-ethereum/log"
-	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/require"
+	"go.uber.org/mock/gomock"
 )
 
 // TestRouting tests that the routes were properly encoded.

store/generated_key/eigenda/eigenda.go

@@ -124,7 +124,9 @@ func (e Store) Put(ctx context.Context, value []byte) ([]byte, error) {
 		return nil, fmt.Errorf("failed to verify commitment: %w", err)
 	}
 
-	err = e.verifier.VerifyCert(ctx, cert)
+	// we set RollupL1InclusionBlockNum to -1 to skip the check, as the cert will only be included
+	// in the batcher's inbox after the proxy returns the verified cert to the batcher.
+	err = e.verifier.VerifyCert(ctx, cert, common.VerifyOptions{RollupL1InclusionBlockNum: -1})
 	if err != nil {
 		return nil, fmt.Errorf("failed to verify DA cert: %w", err)
 	}
@@ -144,7 +146,8 @@ func (e Store) BackendType() common.BackendType {
 
 // Key is used to recover certificate fields and that verifies blob
 // against commitment to ensure data is valid and non-tampered.
-func (e Store) Verify(ctx context.Context, key []byte, value []byte) error {
+// l1InclusionBlockNum is optional and used to validate the certificate: negative number means don't verify this check
+func (e Store) Verify(ctx context.Context, key []byte, value []byte, opts common.VerifyOptions) error {
 	var cert verify.Certificate
 	err := rlp.DecodeBytes(key, &cert)
 	if err != nil {
@@ -164,5 +167,5 @@ func (e Store) Verify(ctx context.Context, key []byte, value []byte) error {
 	}
 
 	// verify DA certificate against EigenDA's batch metadata that's bridged to Ethereum
-	return e.verifier.VerifyCert(ctx, &cert)
+	return e.verifier.VerifyCert(ctx, &cert, opts)
 }

store/generated_key/memstore/memstore.go

@@ -219,7 +219,7 @@ func (e *MemStore) Put(_ context.Context, value []byte) ([]byte, error) {
 	return certBytes, nil
 }
 
-func (e *MemStore) Verify(_ context.Context, _, _ []byte) error {
+func (e *MemStore) Verify(_ context.Context, _, _ []byte, _ common.VerifyOptions) error {
 	return nil
 }
 

store/manager.go

@@ -14,7 +14,9 @@ import (
 
 // IManager ... read/write interface
 type IManager interface {
-	Get(ctx context.Context, key []byte, cm commitments.CommitmentMode) ([]byte, error)
+	// Get fetches a value from a storage backend based on the (commitment mode, type).
+	// It also validates the value retrieved and returns an error if the value is invalid.
+	Get(ctx context.Context, key []byte, cm commitments.CommitmentMode, verifyOpts common.VerifyOptions) ([]byte, error)
 	Put(ctx context.Context, cm commitments.CommitmentMode, key, value []byte) ([]byte, error)
 }
 
@@ -41,7 +43,7 @@ func NewManager(eigenda common.GeneratedKeyStore, s3 common.PrecomputedKeyStore,
 }
 
 // Get ... fetches a value from a storage backend based on the (commitment mode, type)
-func (m *Manager) Get(ctx context.Context, key []byte, cm commitments.CommitmentMode) ([]byte, error) {
+func (m *Manager) Get(ctx context.Context, key []byte, cm commitments.CommitmentMode, verifyOpts common.VerifyOptions) ([]byte, error) {
 	switch cm {
 	case commitments.OptimismKeccak:
 
@@ -71,7 +73,7 @@ func (m *Manager) Get(ctx context.Context, key []byte, cm commitments.Commitment
 		// 1 - read blob from cache if enabled
 		if m.secondary.CachingEnabled() {
 			m.log.Debug("Retrieving data from cached backends")
-			data, err := m.secondary.MultiSourceRead(ctx, key, false, m.eigenda.Verify)
+			data, err := m.secondary.MultiSourceRead(ctx, key, false, m.eigenda.Verify, verifyOpts)
 			if err == nil {
 				return data, nil
 			}
@@ -83,7 +85,7 @@ func (m *Manager) Get(ctx context.Context, key []byte, cm commitments.Commitment
 		data, err := m.eigenda.Get(ctx, key)
 		if err == nil {
 			// verify
-			err = m.eigenda.Verify(ctx, key, data)
+			err = m.eigenda.Verify(ctx, key, data, verifyOpts)
 			if err != nil {
 				return nil, err
 			}
@@ -92,7 +94,7 @@ func (m *Manager) Get(ctx context.Context, key []byte, cm commitments.Commitment
 
 		// 3 - read blob from fallbacks if enabled and data is non-retrievable from EigenDA
 		if m.secondary.FallbackEnabled() {
-			data, err = m.secondary.MultiSourceRead(ctx, key, true, m.eigenda.Verify)
+			data, err = m.secondary.MultiSourceRead(ctx, key, true, m.eigenda.Verify, verifyOpts)
 			if err != nil {
 				m.log.Error("Failed to read from fallback targets", "err", err)
 				return nil, err

store/secondary.go

@@ -29,7 +29,11 @@ type ISecondary interface {
 	CachingEnabled() bool
 	FallbackEnabled() bool
 	HandleRedundantWrites(ctx context.Context, commitment []byte, value []byte) error
-	MultiSourceRead(context.Context, []byte, bool, func(context.Context, []byte, []byte) error) ([]byte, error)
+	// verify fn signature has to match that of common/store.go's GeneratedKeyStore.Verify fn.
+	MultiSourceRead(
+		ctx context.Context, commitment []byte, fallback bool,
+		verify func(context.Context, []byte, []byte, common.VerifyOptions) error, verifyOpts common.VerifyOptions,
+	) ([]byte, error)
 	WriteSubscriptionLoop(ctx context.Context)
 }
 
@@ -142,7 +146,11 @@ func (sm *SecondaryManager) WriteSubscriptionLoop(ctx context.Context) {
 // MultiSourceRead ... reads from a set of backends and returns the first successfully read blob
 // NOTE: - this can also be parallelized when reading from multiple sources and discarding connections that fail
 //   - for complete optimization we can profile secondary storage backends to determine the fastest / most reliable and always rout to it first
-func (sm *SecondaryManager) MultiSourceRead(ctx context.Context, commitment []byte, fallback bool, verify func(context.Context, []byte, []byte) error) ([]byte, error) {
+func (sm *SecondaryManager) MultiSourceRead(
+	ctx context.Context, commitment []byte, fallback bool,
+	// verifyOpts are passed to the verification function
+	verify func(context.Context, []byte, []byte, common.VerifyOptions) error, verifyOpts common.VerifyOptions,
+) ([]byte, error) {
 	var sources []common.PrecomputedKeyStore
 	if fallback {
 		sources = sm.fallbacks
@@ -168,7 +176,7 @@ func (sm *SecondaryManager) MultiSourceRead(ctx context.Context, commitment []by
 
 		// verify cert:data using provided verification function
 		sm.verifyLock.Lock()
-		err = verify(ctx, commitment, data)
+		err = verify(ctx, commitment, data, verifyOpts)
 		if err != nil {
 			cb(Failed)
 			log.Warn("Failed to verify blob", "err", err, "backend", src.BackendType())

verify/cert.go

@@ -54,7 +54,7 @@ func NewCertVerifier(cfg *Config, l log.Logger) (*CertVerifier, error) {
 // verifyBatchConfirmedOnChain verifies that batchMetadata (typically part of a received cert)
 // matches the batch metadata hash stored on-chain
 func (cv *CertVerifier) verifyBatchConfirmedOnChain(
-	ctx context.Context, batchID uint32, batchMetadata *disperser.BatchMetadata,
+	ctx context.Context, batchID uint32, batchMetadata *disperser.BatchMetadata, l1InclusionBlockNumber int64,
 ) error {
 	// 1. Verify batch is actually onchain at the batchMetadata's state confirmedBlockNumber.
 	// This is super unlikely if the disperser is honest, but it could technically happen that a confirmed batch's block gets reorged out,