fix e2e tests

Author: randygrok

da/cmd/local-da/server.go

@@ -2,6 +2,9 @@ package main
 
 import (
 	"context"
+	"crypto/sha256"
+	"encoding/binary"
+	"fmt"
 	"net"
 	"net/http"
 	"sync/atomic"
@@ -13,13 +16,35 @@ import (
 	"github.com/evstack/ev-node/da"
 )
 
+// Blob represents a Celestia-compatible blob for the blob API
+type Blob struct {
+	Namespace  []byte `json:"namespace"`
+	Data       []byte `json:"data"`
+	ShareVer   uint32 `json:"share_version"`
+	Commitment []byte `json:"commitment"`
+	Index      int    `json:"index"`
+}
+
+// Proof represents a Celestia-compatible inclusion proof
+type Proof struct {
+	Data []byte `json:"data"`
+}
+
+// SubmitOptions contains options for blob submission
+type SubmitOptions struct {
+	Fee           float64 `json:"fee,omitempty"`
+	GasLimit      uint64  `json:"gas_limit,omitempty"`
+	SignerAddress string  `json:"signer_address,omitempty"`
+}
+
 // Server is a jsonrpc service that serves the LocalDA implementation
 type Server struct {
 	logger   zerolog.Logger
 	srv      *http.Server
 	rpc      *jsonrpc.RPCServer
 	listener net.Listener
 	daImpl   da.DA
+	localDA  *LocalDA // For blob API access to internal data
 
 	started atomic.Bool
 }
@@ -72,6 +97,184 @@ func (s *serverInternalAPI) SubmitWithOptions(ctx context.Context, blobs []da.Bl
 	return s.daImpl.SubmitWithOptions(ctx, blobs, gasPrice, ns, options)
 }
 
+// blobAPI provides Celestia-compatible Blob API methods
+type blobAPI struct {
+	logger  zerolog.Logger
+	localDA *LocalDA
+}
+
+// Submit submits blobs and returns the DA height (Celestia blob API compatible)
+func (b *blobAPI) Submit(ctx context.Context, blobs []*Blob, opts *SubmitOptions) (uint64, error) {
+	b.logger.Debug().Int("num_blobs", len(blobs)).Msg("blob.Submit called")
+
+	if len(blobs) == 0 {
+		return 0, nil
+	}
+
+	ns := blobs[0].Namespace
+
+	rawBlobs := make([][]byte, len(blobs))
+	for i, blob := range blobs {
+		rawBlobs[i] = blob.Data
+	}
+
+	var gasPrice float64
+	if opts != nil {
+		gasPrice = opts.Fee
+	}
+
+	_, err := b.localDA.Submit(ctx, rawBlobs, gasPrice, ns)
+	if err != nil {
+		return 0, err
+	}
+
+	b.localDA.mu.Lock()
+	height := b.localDA.height
+	b.localDA.mu.Unlock()
+
+	b.logger.Info().Uint64("height", height).Int("num_blobs", len(blobs)).Msg("blob.Submit successful")
+	return height, nil
+}
+
+// Get retrieves a single blob by commitment at a given height (Celestia blob API compatible)
+func (b *blobAPI) Get(ctx context.Context, height uint64, ns []byte, commitment []byte) (*Blob, error) {
+	b.logger.Debug().Uint64("height", height).Msg("blob.Get called")
+
+	blobs, err := b.GetAll(ctx, height, [][]byte{ns})
+	if err != nil {
+		return nil, err
+	}
+
+	for _, blob := range blobs {
+		if len(commitment) == 0 || bytesEqual(blob.Commitment, commitment) {
+			return blob, nil
+		}
+	}
+
+	return nil, nil
+}
+
+// GetAll retrieves all blobs at a given height for the specified namespaces (Celestia blob API compatible)
+func (b *blobAPI) GetAll(ctx context.Context, height uint64, namespaces [][]byte) ([]*Blob, error) {
+	b.logger.Debug().Uint64("height", height).Int("num_namespaces", len(namespaces)).Msg("blob.GetAll called")
+
+	if len(namespaces) == 0 {
+		return []*Blob{}, nil
+	}
+
+	ns := namespaces[0]
+
+	b.localDA.mu.Lock()
+	defer b.localDA.mu.Unlock()
+
+	if height > b.localDA.height {
+		b.logger.Debug().Uint64("requested", height).Uint64("current", b.localDA.height).Msg("blob.GetAll: height in future")
+		return nil, fmt.Errorf("height %d from future, current height is %d", height, b.localDA.height)
+	}
+
+	kvps, ok := b.localDA.data[height]
+	if !ok {
+		b.logger.Debug().Uint64("height", height).Msg("blob.GetAll: no data for height")
+		return []*Blob{}, nil
+	}
+
+	blobs := make([]*Blob, 0, len(kvps))
+	for i, kv := range kvps {
+		var commitment []byte
+		if len(kv.key) > 8 {
+			commitment = kv.key[8:]
+		} else {
+			hash := sha256.Sum256(kv.value)
+			commitment = hash[:]
+		}
+
+		blobs = append(blobs, &Blob{
+			Namespace:  ns,
+			Data:       kv.value,
+			ShareVer:   0,
+			Commitment: commitment,
+			Index:      i,
+		})
+	}
+
+	b.logger.Debug().Uint64("height", height).Int("num_blobs", len(blobs)).Msg("blob.GetAll successful")
+	return blobs, nil
+}
+
+// GetProof retrieves the inclusion proof for a blob (Celestia blob API compatible)
+func (b *blobAPI) GetProof(ctx context.Context, height uint64, ns []byte, commitment []byte) (*Proof, error) {
+	b.logger.Debug().Uint64("height", height).Msg("blob.GetProof called")
+
+	b.localDA.mu.Lock()
+	defer b.localDA.mu.Unlock()
+
+	kvps, ok := b.localDA.data[height]
+	if !ok {
+		return nil, nil
+	}
+
+	for _, kv := range kvps {
+		var blobCommitment []byte
+		if len(kv.key) > 8 {
+			blobCommitment = kv.key[8:]
+		}
+
+		if len(commitment) == 0 || bytesEqual(blobCommitment, commitment) {
+			proof := b.localDA.getProof(kv.key, kv.value)
+			return &Proof{Data: proof}, nil
+		}
+	}
+
+	return nil, nil
+}
+
+// Included checks whether a blob is included in the DA layer (Celestia blob API compatible)
+func (b *blobAPI) Included(ctx context.Context, height uint64, ns []byte, proof *Proof, commitment []byte) (bool, error) {
+	b.logger.Debug().Uint64("height", height).Msg("blob.Included called")
+
+	b.localDA.mu.Lock()
+	defer b.localDA.mu.Unlock()
+
+	kvps, ok := b.localDA.data[height]
+	if !ok {
+		return false, nil
+	}
+
+	for _, kv := range kvps {
+		var blobCommitment []byte
+		if len(kv.key) > 8 {
+			blobCommitment = kv.key[8:]
+		}
+
+		if bytesEqual(blobCommitment, commitment) {
+			return true, nil
+		}
+	}
+
+	return false, nil
+}
+
+// bytesEqual compares two byte slices
+func bytesEqual(a, b []byte) bool {
+	if len(a) != len(b) {
+		return false
+	}
+	for i := range a {
+		if a[i] != b[i] {
+			return false
+		}
+	}
+	return true
+}
+
+// makeID creates an ID from height and commitment
+func makeID(height uint64, commitment []byte) []byte {
+	id := make([]byte, 8+len(commitment))
+	binary.LittleEndian.PutUint64(id, height)
+	copy(id[8:], commitment)
+	return id
+}
+
 func getKnownErrorsMapping() jsonrpc.Errors {
 	errs := jsonrpc.NewErrors()
 	errs.Register(jsonrpc.ErrorCode(da.StatusNotFound), &da.ErrBlobNotFound)
@@ -86,25 +289,35 @@ func getKnownErrorsMapping() jsonrpc.Errors {
 }
 
 // NewServer creates a new JSON-RPC server for the LocalDA implementation
-func NewServer(logger zerolog.Logger, address, port string, daImplementation da.DA) *Server {
+// It registers both the legacy "da" namespace and the Celestia-compatible "blob" namespace
+func NewServer(logger zerolog.Logger, address, port string, localDA *LocalDA) *Server {
 	rpc := jsonrpc.NewServer(jsonrpc.WithServerErrors(getKnownErrorsMapping()))
 	srv := &Server{
-		rpc:    rpc,
-		logger: logger,
-		daImpl: daImplementation,
+		rpc:     rpc,
+		logger:  logger,
+		daImpl:  localDA,
+		localDA: localDA,
 		srv: &http.Server{
 			Addr:              address + ":" + port,
 			ReadHeaderTimeout: 2 * time.Second,
 		},
 	}
 	srv.srv.Handler = http.HandlerFunc(rpc.ServeHTTP)
 
-	apiHandler := &serverInternalAPI{
+	// Register legacy "da" namespace API
+	daAPIHandler := &serverInternalAPI{
 		logger: logger,
-		daImpl: daImplementation,
+		daImpl: localDA,
+	}
+	srv.rpc.Register("da", daAPIHandler)
+
+	// Register Celestia-compatible "blob" namespace API
+	blobAPIHandler := &blobAPI{
+		logger:  logger,
+		localDA: localDA,
 	}
+	srv.rpc.Register("blob", blobAPIHandler)
 
-	srv.rpc.Register("da", apiHandler)
 	return srv
 }
 

test/e2e/evm_test_common.go

@@ -404,6 +404,10 @@ func setupFullNode(t *testing.T, sut *SystemUnderTest, fullNodeHome, sequencerHo
 	err = os.WriteFile(fullNodeGenesis, genesisData, 0644)
 	require.NoError(t, err, "failed to write full node genesis file")
 
+	// Read namespace from sequencer config to pass to full node
+	sequencerConfigPath := filepath.Join(sequencerHome, "config", "evnode.yaml")
+	namespace := extractNamespaceFromConfig(t, sequencerConfigPath)
+
 	// Create JWT secret file for full node
 	fullNodeJwtSecretFile := createJWTSecretFile(t, fullNodeHome, fullNodeJwtSecret)
 
@@ -418,6 +422,7 @@ func setupFullNode(t *testing.T, sut *SystemUnderTest, fullNodeHome, sequencerHo
 		"--evm.eth-url", endpoints.GetFullNodeEthURL(),
 		"--rollkit.da.block_time", DefaultDABlockTime,
 		"--rollkit.da.address", endpoints.GetDAAddress(),
+		"--rollkit.da.namespace", namespace, // Use same namespace as sequencer
 		"--rollkit.rpc.address", endpoints.GetFullNodeRPCListen(),
 		"--rollkit.p2p.listen_address", endpoints.GetFullNodeP2PAddress(),
 	}
@@ -427,6 +432,42 @@ func setupFullNode(t *testing.T, sut *SystemUnderTest, fullNodeHome, sequencerHo
 	sut.AwaitNodeLive(t, endpoints.GetFullNodeRPCAddress(), NodeStartupTimeout)
 }
 
+// extractNamespaceFromConfig reads the namespace from a config file
+func extractNamespaceFromConfig(t *testing.T, configPath string) string {
+	t.Helper()
+
+	configData, err := os.ReadFile(configPath)
+	require.NoError(t, err, "failed to read config file")
+
+	// Parse YAML - look for "namespace:" under "da:" section
+	lines := strings.Split(string(configData), "\n")
+	inDASection := false
+	for _, line := range lines {
+		// Check if we're entering the da: section
+		if strings.TrimSpace(line) == "da:" {
+			inDASection = true
+			continue
+		}
+		// Check if we're leaving the da: section (new top-level key)
+		if inDASection && len(line) > 0 && line[0] != ' ' && line[0] != '\t' {
+			inDASection = false
+		}
+		// Look for namespace: inside the da: section
+		if inDASection {
+			trimmed := strings.TrimSpace(line)
+			if strings.HasPrefix(trimmed, "namespace:") {
+				parts := strings.SplitN(trimmed, ":", 2)
+				if len(parts) == 2 {
+					return strings.TrimSpace(parts[1])
+				}
+			}
+		}
+	}
+
+	t.Fatal("namespace not found in config file")
+	return ""
+}
+
 // Global nonce counter to ensure unique nonces across multiple transaction submissions
 var globalNonce uint64 = 0